[Kafka API] Add forwarding of txn requests to coordinator (ydb-platform#17765)

Author: a-serebryanskiy

ydb/core/kafka_proxy/kafka_connection.cpp

@@ -340,6 +340,42 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
         Register(CreateKafkaAlterConfigsActor(Context, header->CorrelationId, message));
     }
 
+    void HandleMessage(const TRequestHeaderData* header, const TMessagePtr<TAddPartitionsToTxnRequestData>& message) {
+        Send(MakeTransactionsServiceID(), new TEvKafka::TEvAddPartitionsToTxnRequest(
+            header->CorrelationId, 
+            message,
+            Context->ConnectionId,
+            Context->DatabasePath
+        ));
+    }
+
+    void HandleMessage(const TRequestHeaderData* header, const TMessagePtr<TAddOffsetsToTxnRequestData>& message) {
+        Send(MakeTransactionsServiceID(), new TEvKafka::TEvAddOffsetsToTxnRequest(
+            header->CorrelationId, 
+            message,
+            Context->ConnectionId,
+            Context->DatabasePath
+        ));
+    }
+
+    void HandleMessage(const TRequestHeaderData* header, const TMessagePtr<TTxnOffsetCommitRequestData>& message) {
+        Send(MakeTransactionsServiceID(), new TEvKafka::TEvTxnOffsetCommitRequest(
+            header->CorrelationId, 
+            message,
+            Context->ConnectionId,
+            Context->DatabasePath
+        ));
+    }
+
+    void HandleMessage(const TRequestHeaderData* header, const TMessagePtr<TEndTxnRequestData>& message) {
+        Send(MakeTransactionsServiceID(), new TEvKafka::TEvEndTxnRequest(
+            header->CorrelationId, 
+            message,
+            Context->ConnectionId,
+            Context->DatabasePath
+        ));
+    }
+
     template<class T>
     TMessagePtr<T> Cast(std::shared_ptr<Msg>& request) {
         return TMessagePtr<T>(request->Buffer, request->Message);
@@ -443,6 +479,22 @@ class TKafkaConnection: public TActorBootstrapped<TKafkaConnection>, public TNet
                 HandleMessage(&Request->Header, Cast<TAlterConfigsRequestData>(Request));
                 break;
 
+            case ADD_PARTITIONS_TO_TXN:
+                HandleMessage(&Request->Header, Cast<TAddPartitionsToTxnRequestData>(Request));
+                break;
+
+            case ADD_OFFSETS_TO_TXN:
+                HandleMessage(&Request->Header, Cast<TAddOffsetsToTxnRequestData>(Request));
+                break;
+
+            case TXN_OFFSET_COMMIT:
+                HandleMessage(&Request->Header, Cast<TTxnOffsetCommitRequestData>(Request));
+                break;
+
+            case END_TXN:
+                HandleMessage(&Request->Header, Cast<TEndTxnRequestData>(Request));
+                break;
+
             default:
                 KAFKA_LOG_ERROR("Unsupported message: ApiKey=" << Request->Header.RequestApiKey);
                 PassAway();