Fix reading messages from Kafka API in no authentication mode (#13357)

Author: a-serebryanskiy

ydb/core/kafka_proxy/actors/actors.h

@@ -51,7 +51,9 @@ struct TContext {
 
     NKikimr::NPQ::TRlContext RlContext;
 
-    bool Authenticated() { return AuthenticationStep == SUCCESS; }
+    bool Authenticated() { 
+        return !RequireAuthentication || AuthenticationStep == SUCCESS; 
+    }
 };
 
 template<std::derived_from<TApiMessage> T>

ydb/core/kafka_proxy/kafka_connection.cpp

@@ -339,7 +339,7 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
         if (Request->Header.ClientId.has_value() && Request->Header.ClientId != "") {
             Context->KafkaClient = Request->Header.ClientId.value();
         }
-        
+
         switch (Request->Header.RequestApiKey) {
             case PRODUCE:
                 HandleMessage(&Request->Header, Cast<TProduceRequestData>(Request), ctx);
@@ -627,10 +627,12 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
                     case INFLIGTH_CHECK:
                         if (!Context->Authenticated() && !PendingRequestsQueue.empty()) {
                             // Allow only one message to be processed at a time for non-authenticated users
+                            KAFKA_LOG_ERROR("DoRead: failed inflight check: there are " << PendingRequestsQueue.size() << " pending requests and user is not authnicated.  Only one paraller request is allowed for a non-authenticated user.");
                             return true;
                         }
                         if (InflightSize + Request->ExpectedSize > Context->Config.GetMaxInflightSize()) {
                             // We limit the size of processed messages so as not to exceed the size of available memory
+                            KAFKA_LOG_ERROR("DoRead: failed inflight check: InflightSize + Request->ExpectedSize=" << InflightSize + Request->ExpectedSize << " > Context->Config.GetMaxInflightSize=" << Context->Config.GetMaxInflightSize());
                             return true;
                         }
                         InflightSize += Request->ExpectedSize;
@@ -713,12 +715,9 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
     }
 
     bool RequireAuthentication(EApiKey apiKey) {
-        bool configuredToAuthenticate = NKikimr::AppData()->EnforceUserTokenRequirement;
-        bool apiKeyRequiresAuthentication = !(EApiKey::API_VERSIONS == apiKey || 
-                                              EApiKey::SASL_HANDSHAKE == apiKey || 
-                                              EApiKey::SASL_AUTHENTICATE == apiKey);
-                                            
-        return configuredToAuthenticate && apiKeyRequiresAuthentication;
+        return !(EApiKey::API_VERSIONS == apiKey || 
+                EApiKey::SASL_HANDSHAKE == apiKey || 
+                EApiKey::SASL_AUTHENTICATE == apiKey);
     }
 
     void HandleConnected(TEvPollerReady::TPtr event, const TActorContext& ctx) {