fix

Author: niksaveliev

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

@@ -50,7 +50,7 @@ static EKafkaErrors KqpStatusToKafkaError(Ydb::StatusIds::StatusCode status) {
     } else if (status == Ydb::StatusIds::PRECONDITION_FAILED) {
         return EKafkaErrors::REBALANCE_IN_PROGRESS;
     }
-    return EKafkaErrors::UNKNOWN_SERVER_ERROR;
+    return EKafkaErrors::INVALID_REQUEST;
 }
 
 void TKafkaBalancerActor::Handle(TEvents::TEvWakeup::TPtr&, const TActorContext& ctx) {

ydb/core/kafka_proxy/kafka_connection.cpp

@@ -244,12 +244,10 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
             HandleKillReadSession();
             Register(new TKafkaBalancerActor(Context, 0, header->CorrelationId, message));
         } else {
-            if (ReadSessionActorId) {
-                Send(ReadSessionActorId, new TEvKafka::TEvJoinGroupRequest(header->CorrelationId, message));
-            } else {
+            if (!ReadSessionActorId) {
                 ReadSessionActorId = RegisterWithSameMailbox(CreateKafkaReadSessionActor(Context, 0));
-                Send(ReadSessionActorId, new TEvKafka::TEvJoinGroupRequest(header->CorrelationId, message));
             }
+            Send(ReadSessionActorId, new TEvKafka::TEvJoinGroupRequest(header->CorrelationId, message));
         }
     }
 
@@ -258,12 +256,10 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
             HandleKillReadSession();
             Register(new TKafkaBalancerActor(Context, 0, header->CorrelationId, message));
         } else {
-            if (ReadSessionActorId) {
-                Send(ReadSessionActorId, new TEvKafka::TEvSyncGroupRequest(header->CorrelationId, message));
-            } else {
+            if (!ReadSessionActorId) {
                 ReadSessionActorId = RegisterWithSameMailbox(CreateKafkaReadSessionActor(Context, 0));
-                Send(ReadSessionActorId, new TEvKafka::TEvSyncGroupRequest(header->CorrelationId, message));
             }
+            Send(ReadSessionActorId, new TEvKafka::TEvSyncGroupRequest(header->CorrelationId, message));
         }
     }
 
@@ -272,12 +268,10 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
             HandleKillReadSession();
             Register(new TKafkaBalancerActor(Context, 0, header->CorrelationId, message));
         } else {
-            if (ReadSessionActorId) {
-                Send(ReadSessionActorId, new TEvKafka::TEvHeartbeatRequest(header->CorrelationId, message));
-            } else {
+            if (!ReadSessionActorId) {
                 ReadSessionActorId = RegisterWithSameMailbox(CreateKafkaReadSessionActor(Context, 0));
-                Send(ReadSessionActorId, new TEvKafka::TEvHeartbeatRequest(header->CorrelationId, message));
             }
+            Send(ReadSessionActorId, new TEvKafka::TEvHeartbeatRequest(header->CorrelationId, message));
         }
     }
 
@@ -286,12 +280,10 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
             HandleKillReadSession();
             Register(new TKafkaBalancerActor(Context, 0, header->CorrelationId, message));
         } else {
-            if (ReadSessionActorId) {
-                Send(ReadSessionActorId, new TEvKafka::TEvLeaveGroupRequest(header->CorrelationId, message));
-            } else {
+            if (!ReadSessionActorId) {
                 ReadSessionActorId = RegisterWithSameMailbox(CreateKafkaReadSessionActor(Context, 0));
-                Send(ReadSessionActorId, new TEvKafka::TEvLeaveGroupRequest(header->CorrelationId, message));
             }
+            Send(ReadSessionActorId, new TEvKafka::TEvLeaveGroupRequest(header->CorrelationId, message));
         }
     }