Kafka auth flag and bugfix in ts (#14220)

Author: FloatingCrowbar

ydb/core/kafka_proxy/actors/kafka_fetch_actor.cpp

@@ -148,7 +148,7 @@ void TKafkaFetchActor::FillRecordsBatch(const NKikimrClient::TPersQueueFetchResp
         }
 
         lastOffset = result.GetOffset();
-        lastTimestamp = result.GetWriteTimestampMS();
+        lastTimestamp = result.GetCreateTimestampMS();
         auto& record = recordsBatch.Records[recordIndex];
 
         record.DataChunk = NKikimr::GetDeserializedData(result.GetData());

ydb/core/kafka_proxy/actors/kafka_sasl_auth_actor.cpp

@@ -161,7 +161,12 @@ void TKafkaSaslAuthActor::SendLoginRequest(TKafkaSaslAuthActor::TAuthData authDa
 
 void TKafkaSaslAuthActor::SendApiKeyRequest() {
     auto entries = NKikimr::NGRpcProxy::V1::GetTicketParserEntries(DatabaseId, FolderId);
-
+    TString ticket;
+    if (Context->Config.GetAuthViaApiKey()) {
+        ticket = "ApiKey " + ClientAuthData.Password;
+    } else {
+        ticket = ClientAuthData.Password;
+    }
     Send(NKikimr::MakeTicketParserID(), new NKikimr::TEvTicketParser::TEvAuthorizeTicket({
         .Database = DatabasePath,
         .Ticket = "ApiKey " + ClientAuthData.Password,

ydb/core/protos/config.proto

@@ -2125,7 +2125,8 @@ message TKafkaProxyConfig {
     }
 
     optional TProxy Proxy = 7;
-    optional bool MeteringV2Enabled = 10 [default = false];
+    optional bool MeteringV2Enabled = 10 [default = true];
+    optional bool AuthViaApiKey = 11 [default = true];
 }
 
 message TAwsCompatibilityConfig {