Kafka connect distributed #12960

niksaveliev · 2024-12-25T00:53:48Z

No description provided.

github-actions · 2024-12-25T00:57:29Z

⚪ 2024-12-25 00:57:28 UTC Pre-commit check linux-x86_64-release-asan for ee21fbe has started.
⚪ 2024-12-25 00:57:41 UTC Artifacts will be uploaded here
⚪ 2024-12-25 01:00:09 UTC ya make is running...
🟢 2024-12-25 01:00:15 UTC Tests successful.

Test history | Ya make output | Test bloat

TESTS	PASSED	ERRORS	FAILED	SKIPPED	MUTED^?
0	0	0	0	0	0

🟢 2024-12-25 01:00:20 UTC Build successful.

github-actions · 2024-12-25T00:58:12Z

⚪ 2024-12-25 00:58:11 UTC Pre-commit check linux-x86_64-relwithdebinfo for ee21fbe has started.
⚪ 2024-12-25 00:58:22 UTC Artifacts will be uploaded here
⚪ 2024-12-25 01:00:48 UTC ya make is running...
🟢 2024-12-25 01:00:53 UTC Tests successful.

Test history | Ya make output | Test bloat

TESTS	PASSED	ERRORS	FAILED	SKIPPED	MUTED^?
0	0	0	0	0	0

🟢 2024-12-25 01:00:59 UTC Build successful.

github-actions · 2024-12-25T07:24:16Z

⚪ 2024-12-25 07:24:15 UTC Pre-commit check linux-x86_64-release-asan for e106f83 has started.
⚪ 2024-12-25 07:24:28 UTC Artifacts will be uploaded here
⚪ 2024-12-25 07:26:58 UTC ya make is running...
🟢 2024-12-25 07:27:04 UTC Tests successful.

Test history | Ya make output | Test bloat

TESTS	PASSED	ERRORS	FAILED	SKIPPED	MUTED^?
0	0	0	0	0	0

🟢 2024-12-25 07:27:09 UTC Build successful.

github-actions · 2024-12-25T07:26:28Z

⚪ 2024-12-25 07:26:28 UTC Pre-commit check linux-x86_64-relwithdebinfo for e106f83 has started.
⚪ 2024-12-25 07:26:39 UTC Artifacts will be uploaded here
⚪ 2024-12-25 07:28:59 UTC ya make is running...
🟢 2024-12-25 07:29:05 UTC Tests successful.

Test history | Ya make output | Test bloat

TESTS	PASSED	ERRORS	FAILED	SKIPPED	MUTED^?
0	0	0	0	0	0

🟢 2024-12-25 07:29:10 UTC Build successful.

alexnick88 · 2024-12-25T08:00:49Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+
+
+void TKafkaBalancerActor::Handle(NKqp::TEvKqp::TEvCreateSessionResponse::TPtr& ev, const TActorContext& ctx) {
+    const TString createSessionError = "Failed to create KQP session";


get error text also from Issues inside response.

alexnick88 · 2024-12-25T08:01:04Z

ydb/core/kafka_proxy/actors/kqp_balance_transaction.cpp

+bool TKqpTxHelper::HandleCreateSessionResponse(NKqp::TEvKqp::TEvCreateSessionResponse::TPtr& ev, const TActorContext&) {
+    const auto& record = ev->Get()->Record;
+
+    if (record.GetYdbStatus() != Ydb::StatusIds::SUCCESS) {


Add logs here

alexnick88 · 2024-12-25T08:02:07Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+void TKafkaBalancerActor::Handle(NKqp::TEvKqp::TEvQueryResponse::TPtr& ev, const TActorContext& ctx) {
+    const TString kqpQueryError = "KQP query error";
+    if (ev->Cookie != KqpReqCookie) {
+        KAFKA_LOG_CRIT("Unexpected cookie in TEvQueryResponse");


Really crit? Is this situation possible?

alexnick88 · 2024-12-25T08:02:54Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+
+    const auto& record = ev->Get()->Record;
+    auto status   = record.GetYdbStatus();
+    auto kafkaErr = KqpStatusToKafkaError(status);


Fill kqpQueryError from status issues and status

alexnick88 · 2024-12-25T08:03:51Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+    if (kafkaErr != EKafkaErrors::NONE_ERROR) {
+        switch (RequestType) {
+            case JOIN_GROUP:
+                SendJoinGroupResponseFail(ctx, CorellationId, kafkaErr, kqpQueryError);


Refactor this code. Make SendResponseFail() function and switch with RequestType inside. Otherwise there is copy-paste.

alexnick88 · 2024-12-25T08:07:46Z

ydb/core/kafka_proxy/actors/kafka_balance_actor_sql.cpp

+    LIMIT 1;
+)";
+
+const TString INSERT_MEMBER_AND_SELECT_MASTER = R"(


In code you using INSERT_MEMBER_AND_SELECT_MASTER_QUERY. No _QUERY suffix here.
And you should check that visibility of transaction changes is enabled. What if min join-time in this request?

alexnick88 · 2024-12-25T08:08:24Z

ydb/core/kafka_proxy/actors/kafka_balance_actor_sql.cpp

+        $Generation,
+        $MemberId,
+        CurrentUtcDateTime(),
+        CurrentUtcDateTime() + Interval("PT5S"),


WTF? Make all dates parameters. No 5 seconds in YQL.

alexnick88 · 2024-12-25T08:18:34Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+            if (IsMaster) {
+                auto wakeup = std::make_unique<TEvents::TEvWakeup>(0);
+                ctx.ActorSystem()->Schedule(
+                    TDuration::Seconds(WAKE_UP_DELAY_SECONDS),


Why constant? Get this from client.

alexnick88 · 2024-12-25T08:20:15Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+
+void TKafkaBalancerActor::Handle(NKqp::TEvKqp::TEvCreateSessionResponse::TPtr& ev, const TActorContext& ctx) {
+    const TString createSessionError = "Failed to create KQP session";
+    if (!Kqp->HandleCreateSessionResponse(ev, ctx)) {


if Error is TLI may be you should restart action from begin? However response with error is possible solution. Make an issue for this.

github-actions · 2025-02-05T09:11:49Z

🔴 Unable to merge your PR into the base branch. Please rebase or merge it with the base branch.

alexnick88 · 2025-02-05T12:07:18Z

ydb/core/kafka_proxy/actors/kafka_balance_actor_sql.cpp

+    DECLARE $Database AS Utf8;
+    DECLARE $Master AS Utf8;
+
+    INSERT INTO `/Root/.metadata/kafka_consumer_groups`


Change /Root to correct database or remove it. Check how this is done in WriteSession in choosing partition for write.

alexnick88 · 2025-02-05T12:07:44Z

ydb/core/kafka_proxy/actors/kafka_balance_actor_sql.cpp

+        $Generation,
+        $State,
+        $Database,
+        CurrentUtcDateTime(),


Move Timestamp into parameters please.

alexnick88 · 2025-02-05T12:08:11Z

ydb/core/kafka_proxy/actors/kafka_balance_actor_sql.cpp

+    SET
+        state = $State,
+        generation = $Generation,
+        last_heartbeat_time = CurrentUtcDateTime(),


All values must be in parameters.

a-serebryanskiy · 2025-02-05T13:42:22Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+    if (status == Ydb::StatusIds::SUCCESS) {
+        return EKafkaErrors::NONE_ERROR;
+    }
+    return EKafkaErrors::UNKNOWN_SERVER_ERROR;


А тут еще будут доработки? Или такие ошибки невозможно протранслировать в какие-то конкретные ощибки кафки?

a-serebryanskiy · 2025-02-05T13:43:36Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+
+void TKafkaBalancerActor::Handle(NMetadata::NProvider::TEvManagerPrepared::TPtr&, const TActorContext& ctx) {
+    TablesInited++;
+    if (TablesInited == 2) {


Magic number here. Please change to constant value to improve code readability.

a-serebryanskiy · 2025-02-05T13:59:26Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+            HandleHeartbeatResponse(ev, ctx);
+            break;
+        default:
+            KAFKA_LOG_CRIT("Unknown RequestType in TEvCreateSessionResponse");


May be we should add this log line in every other switch case? To catch a situation, when we add new request type but forget to add it in some switch clause.

a-serebryanskiy · 2025-02-05T14:04:41Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+    auto& record = ev->Get()->Record;
+    auto& resp = record.GetResponse();
+    if (resp.GetYdbResults().empty()) {
+        outGroupExists = false;


May be return complex object with outGroupExists, outGeneration, outState, outMasterId and outTtl fields instead of affecting arguments? I feel like it is an antipattern (in java sure it is, but in c++ i'm not sure). Ready to discuss)

This new returning struct can have some meaningful name like GroupStatus (if i understood code correctly). It will vastly improve readability.

I found myself searching through code for what does state variable mean in this context and I couldn't find it. This new domain object (say GroupStatus) could also have comments for every field to simplify code search.

a-serebryanskiy · 2025-02-05T14:07:13Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+
+bool TKafkaBalancerActor::ParseAssignments(
+    NKqp::TEvKqp::TEvQueryResponse::TPtr ev,
+    TString& assignments)


Same. I'd suggest returning this assginemnts string as a return value of this function. Ready to discuss)

a-serebryanskiy · 2025-02-05T15:30:06Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+            member.MemberId = mId;
+            member.MetaStr = meta;
+            member.Metadata = member.MetaStr;
+            TBuffer buffer(member.Metadata.value().data() + sizeof(TKafkaVersion), member.Metadata.value().size_bytes() - sizeof(TKafkaVersion));


How this buffer is used here?

a-serebryanskiy · 2025-02-06T09:04:36Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+            if (!ParseCheckStateAndGeneration(ev, groupExists, generation, state, masterId, groupTtl) ||
+                !groupExists || generation != GenerationId || state != GROUP_STATE_SYNC) { //
+                SendSyncGroupResponseFail(ctx, CorrelationId,
+                                        EKafkaErrors::UNKNOWN_SERVER_ERROR,


May be INVALID_REQUEST is better here cause its error suggests user to check broker logs.

a-serebryanskiy · 2025-02-06T09:07:28Z

ydb/core/kafka_proxy/actors/kafka_balancer_actor.cpp

+                KqpReqCookie++;
+                NYdb::TParamsBuilder params;
+                params.AddParam("$ConsumerGroup").Utf8(GroupId).Build();
+                params.AddParam("$Database").Utf8(Kqp->DataBase).Build();
+                params.AddParam("$Generation").Uint64(GenerationId).Build();
+                params.AddParam("$State").Uint64(GROUP_STATE_WORKING).Build();
+
+                if (SyncGroupRequestData->Assignments.size() == 0) {
+                    SendSyncGroupResponseFail(ctx, CorrelationId, EKafkaErrors::INVALID_REQUEST);
+                    PassAway();
+                    return;
+                }
+
+                auto& assignmentList = params.AddParam("$Assignments").BeginList();
+                for (auto& assignment: SyncGroupRequestData->Assignments) {
+
+
+                    assignmentList.AddListItem()
+                        .BeginStruct()
+                            .AddMember("MemberId").Utf8(assignment.MemberId.value())
+                            .AddMember("Assignment").String(TString(assignment.Assignment.value().data(),
+                                    assignment.Assignment.value().size()))
+                        .EndStruct();
+                }
+                assignmentList.EndList().Build();
+                Kqp->SendYqlRequest(UPSERT_ASSIGNMENTS_AND_SET_WORKING_STATE, params.Build(), KqpReqCookie, ctx);


I'd suggest extracting building of KQP request and sending in to KQP to the separate method (like SendUpsertAssignmentsAndSetWorkingStateKqpRequest) to improve readability.

a-serebryanskiy · 2025-02-06T09:37:15Z

ydb/core/kafka_proxy/actors/kafka_read_session_actor.h

     */

+static const TString SUPPORTED_ASSIGN_STRATEGY = "roundrobin";


comment not to forget to deprecate this const before commit (or to support feature flag enabled migration)

a-serebryanskiy · 2025-02-06T09:44:00Z

ydb/core/kafka_proxy/kafka_connection.cpp

-            ReadSessionActorId = ctx.RegisterWithSameMailbox(CreateKafkaReadSessionActor(Context, 0));
+
+    void HandleMessage(const TRequestHeaderData* header, const TMessagePtr<TJoinGroupRequestData>& message, const TActorContext& /*ctx*/) {
+        if (ReadSessionActorId) {


comment not to forget to add feature flag dependency here

draft

d015d16

fix

b03fd8b

alexnick88 reviewed Dec 25, 2024

View reviewed changes

work

77ab3d7

fix

48b56a4

alexnick88 reviewed Feb 5, 2025

View reviewed changes

a-serebryanskiy reviewed Feb 6, 2025

View reviewed changes

move db and timestamps to variables

ccba17a

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka connect distributed #12960

Kafka connect distributed #12960

niksaveliev commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

alexnick88 Dec 25, 2024

github-actions bot commented Feb 5, 2025

alexnick88 Feb 5, 2025

alexnick88 Feb 5, 2025

alexnick88 Feb 5, 2025

a-serebryanskiy Feb 5, 2025

a-serebryanskiy Feb 5, 2025

a-serebryanskiy Feb 5, 2025

a-serebryanskiy Feb 5, 2025

a-serebryanskiy Feb 5, 2025

a-serebryanskiy Feb 5, 2025

a-serebryanskiy Feb 5, 2025

a-serebryanskiy Feb 6, 2025

a-serebryanskiy Feb 6, 2025

a-serebryanskiy Feb 6, 2025

a-serebryanskiy Feb 6, 2025



		void TKafkaBalancerActor::Handle(NKqp::TEvKqp::TEvCreateSessionResponse::TPtr& ev, const TActorContext& ctx) {
		const TString createSessionError = "Failed to create KQP session";

		*/

		static const TString SUPPORTED_ASSIGN_STRATEGY = "roundrobin";

Kafka connect distributed #12960

Are you sure you want to change the base?

Kafka connect distributed #12960

Conversation

niksaveliev commented Dec 25, 2024 • edited Loading

github-actions bot commented Dec 25, 2024 • edited Loading

github-actions bot commented Dec 25, 2024 • edited Loading

github-actions bot commented Dec 25, 2024 • edited Loading

github-actions bot commented Dec 25, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Feb 5, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

niksaveliev commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading

github-actions bot commented Dec 25, 2024 •

edited

Loading