[KIP-848] Mock handler and Integration tests passing #4662

emasab · 2024-03-26T17:40:27Z

Mock handler implementation
Rename current consumer protocol from generic to classic
Mock handler with automatic or manual assignment
More consumer group metadata getters
Test helpers
Expedite next HB after FindCoordinator
doing it with an exponential backoff to avoid tight loops
Configurable session timeout and HB interval
Fix mock handler ListOffsets response
LeaderEpoch instead of CurrentLeaderEpoch
Integration tests passing with AK trunk
Improve documentation and KIP 848 specific mock tests

pranavrth

For src files.

.semaphore/semaphore.yml

src/rdkafka.h

pranavrth · 2024-07-15T13:59:25Z

src/rdkafka.h

+/**
+ * @brief Get member epoch of a group metadata.
+ *        Corresponds to the generation id in consumer protocol classic;
+ *
+ * @param group_metadata The group metadata
+ *
+ * @returns The member epoch id contained in the passed \p group_metadata.
+ */
+RD_EXPORT
+int32_t rd_kafka_consumer_group_metadata_member_epoch(
+    const rd_kafka_consumer_group_metadata_t *group_metadata);


Do we have method for retrieving geenration id? If not, do we need one for member epoch? If we really need this for member epoch, let's create one for generation id separately.

In Java GroupMetadata they haven't added a new one for member epoch, maybe naming can be changed in a future major release.

I was saying that shall we add a new method rd_kafka_consumer_group_metadata_generation_id as well.

OK, I think I'll leave only rd_kafka_consumer_group_metadata_generation_id as they didn't add the memberEpoch method in Java

/** * @brief Get the generation id (classic protocol) * or member epoch (consumer protocol) of a group

Is this fine description?

Yes, correct

src/rdkafka_mock.h

pranavrth · 2024-07-17T10:21:09Z

src/rdkafka_mock_cgrp.c

+                if (!RD_KAFKAP_STR_LEN(MemberId)) {
+                        /* Generate a member id */
+                        rd_kafka_Uuid_t member_id = rd_kafka_Uuid_random();
+                        member->id =
+                            rd_strdup(rd_kafka_Uuid_base64str(&member_id));
+                } else
+                        member->id = RD_KAFKAP_STR_DUP(MemberId);


This is not supported by the real broker that we use the user provided MemberId.

I think it needs to be kept because when broker assigns a MemberId and then member is fenced then it needs to rejoin with same MemberId as KIP says It is used in all the communication with the group coordinator and must be kept during the entirely life span of the member (e.g. the consumer).. So even if broker forgot of the member it must accept the MemberId it sends. We have to see if this is possible with the real implementation too with David.

do_test_consumer_group_heartbeat_fenced_error:431: UNKNOWN_MEMBER_ID, variation 1 is failing without this.

I think it should be possible with static membership. We can keep it but lets make sure that we are replicating the real broker behaviour.

Yes, we confirmed in the call that the broker is sometimes accepting the member id from the client, when member is fenced and forgotten but still has a member id given from the broker previously.

src/rdkafka_mock_cgrp.c

pranavrth · 2024-07-17T10:27:44Z

src/rdkafka_mock_cgrp.c

+static void rd_kafka_mock_cgrp_consumer_session_tmr_cb(rd_kafka_timers_t *rkts,
+                                                       void *arg) {
+        rd_kafka_mock_cgrp_consumer_t *mcgrp = arg;
+        rd_kafka_mock_cgrp_consumer_member_t *member, *tmp;
+        rd_ts_t now                       = rd_clock();
+        rd_kafka_mock_cluster_t *mcluster = mcgrp->cluster;
+
+        mtx_unlock(&mcluster->lock);
+        TAILQ_FOREACH_SAFE(member, &mcgrp->members, link, tmp) {
+                if (member->ts_last_activity +
+                        (mcgrp->session_timeout_ms * 1000) >
+                    now)
+                        continue;
+
+                rd_kafka_dbg(mcgrp->cluster->rk, MOCK, "MOCK",
+                             "Member %s session timed out for group %s",
+                             member->id, mcgrp->id);
+
+                rd_kafka_mock_cgrp_consumer_member_fenced(mcgrp, member);
+        }
+        mtx_unlock(&mcluster->lock);
+}


Where do we send session timeout error?

I am still not clear on where are we sending the actual error from the broker to the client. I think we can discuss this in a call.

The session times out on the broker, then member is removed, then member receives FENCED_MEMBER_EPOCH error on next HB

pranavrth · 2024-07-17T10:29:19Z

src/rdkafka_mock_cgrp.c

+        mtx_unlock(&mcluster->lock);
+}
+
+void rd_kafka_mock_set_default_session_timeout(


Why default instead of normal session timeout?

I renamed it to rd_kafka_mock_group_consumer_session_timeout_ms and rd_kafka_mock_group_consumer_heartbeat_interval_ms following the convention introduced with rd_kafka_mock_group_initial_rebalance_delay_ms of rd_kafka_mock + configuration name

We should add _set_. rd_kafka_mock_set_group_consumer_session_timeout_ms

ok even if different from rd_kafka_mock_group_initial_rebalance_delay_ms but avoids collisions with the getter that should be without get

pranavrth · 2024-07-17T10:29:28Z

src/rdkafka_mock_cgrp.c

+        mtx_unlock(&mcluster->lock);
+}
+
+void rd_kafka_mock_set_default_heartbeat_interval(


src/rdkafka_mock_handlers.c

pranavrth · 2024-07-17T14:06:34Z

src/rdkafka_mock_handlers.c

+        rd_kafka_buf_read_str(rkbuf, &ServerAssignor);
+
+        /* #TopicPartitions */
+        rd_kafka_buf_read_arraycnt(rkbuf, &TopicPartitionsCnt,


I think we should use rd_kafka_buf_read_topic_partitions here.

If you do that, then you need to scan the topic partition list, and for each topic partition you need to compare topic id to see if it has changed to call rd_kafka_mock_topic_find_by_id. If it returned like a Map<Topic, List<Partition>> or we didn't have to do an operation at each topic change I'd be more inclined to do than. What's your view on this?

I think its better to use 2 loops here. First to read the data from the rpc and second for our business logic. This ensures readability, reusability and separation of concern.

src/rdkafka_mock_handlers.c

pranavrth

Some 14 files more remaining but the main one is tests/0147-consumer_group_consumer_mock.c.

tests/0045-subscribe_update.c

pranavrth · 2024-07-18T13:30:39Z

tests/0081-admin.c

-        do_test_CreateTopics("temp queue, op timeout 0", rk, NULL, 0, 0);
+        /* FIXME: KRaft async CreateTopics is working differently than
+         * wth Zookeeper
+         * do_test_CreateTopics("temp queue, op timeout 0", rk, NULL, 0, 0); */


Does AK team know about it?

Let's have another function to determine whether its a zookeeper or kraft cluster and disable only when it kraft.

Sure, I opened this issue.
Then talked about it internally. I think it was a bug and not a feature given they didn't acknowledge it was a feature to replicate with KRaft, we could remove the test for both when that is closed.

tests/0081-admin.c

tests/0112-assign_unknown_part.c

pranavrth · 2024-07-18T13:52:12Z

tests/0113-cooperative_rebalance.cpp

+      if (test_consumer_group_protocol_classic() &&
          !(rebalance_cb1.assign_call_cnt == expected_cb1_assign_call_cnt &&
            rebalance_cb2.assign_call_cnt == expected_cb2_assign_call_cnt))


We should have similar calculations for the new protocol as well.

Same for others as well below.

With KIP-848 the callback count isn't fixed as in classic protocol so we can only verify the expected final assignment but not the number of callbacks.

I think we can do that deterministically but might be more complex.

tests/0113-cooperative_rebalance.cpp

pranavrth · 2024-07-18T14:12:59Z

tests/0113-cooperative_rebalance.cpp

+  } else if (0) {
+    /* FIXME: enable this once new errors are handled in OffsetCommit. */
+    t_consumer_group_consumer_retry_commit_on_fenced_member_epoch();
+  }


I will enable this in my other opened PR.

I checked it and I think we should remove this test, as now OffsetCommit returns STALE_MEMBER_EPOCH that is not retried automatically and FENCED _MEMBER_EPOCH is returned by the Heartbeat only

tests/0120-asymmetric_subscription.c

pranavrth · 2024-07-18T14:18:59Z

tests/0143-exponential_backoff_mock.c

+        if (test_consumer_group_protocol_classic()) {
+                ApiKey = RD_KAFKAP_Heartbeat;
+        } else {
+                ApiKey = RD_KAFKAP_ConsumerGroupHeartbeat;
+        }


This should be opposite condition. if condition should be for consumer as that is the new one.

I've always used always test_consumer_group_protocol_classic(), even if implementing the other function (that I can remove) so it's consistent for search, also it isolates the parts that we want to remove in a distant future.
Also if there was a third protocol, we hope not, that should start from the assumptions and tests of the consumer protocol, not with those of the classic protocol.

We should be using test_consumer_group_protocol_consumer at all the places instead of test_consumer_group_protocol_classic. Reason behind this is:

Let's say we add a new protocol in future. We want all the tests to run. We are making sure of that by using !test_consumer_group_protocol_consumer() here instead.

In !test_consumer_group_protocol_classic() condition, we will skip those tests with the new protocol added.

No, in this case we're running the RD_KAFKAP_ConsumerGroupHeartbeat test that is more appropriate than running the classic one, if we use test_consumer_group_protocol_consumer ... else in case of a new protocol we run the classic tests, while running the newer ones would be more appropriate.

I am talking about all the places. Not just here. Basically the idea is to make sure that we run all the test cases when we add new protocols. Let's ensure that when there is a new protocol, all the tests are executed.

yes, all tests about the new protocol, the tests about the classic protocol don't have to be executed as they won't be applicable. Probably the tests about the 848 protocol won't be applicable either but it's more probable it'll be similar to 848 in case it happens.

pranavrth

Some more comments

tests/0016-client_swname.c

tests/0029-assign_offset.c

tests/java/Makefile

pranavrth · 2024-07-25T10:15:28Z

tests/test.c

+        va_start(ap, member_cnt);
+        for (i = 0; i < member_cnt; i++) {
+                rd_kafka_consumer_group_metadata_t *cgmetadata = NULL;
+                rd_kafka_t *c = va_arg(ap, rd_kafka_t *);


better consumer_rk instead of c

Let's use just consumer ok?

tests/test.c

pranavrth · 2024-07-25T10:33:50Z

tests/trivup/trivup-0.12.6.tar.gz

We should release new trivup version. Lets check with Magnus.

tests/testcpp.cpp

tests/0118-commit_rebalance.c

confluent-cla-assistant · 2024-10-03T19:00:53Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

pranavrth · 2024-11-14T10:10:20Z

src/rdkafka_mock_int.h

+        rd_kafka_topic_partition_list_t *
+            target_assignment; /**< Target assignment,


Suggested change

rd_kafka_topic_partition_list_t *

target_assignment; /**< Target assignment,

rd_kafka_topic_partition_list_t

*target_assignment; /**< Target assignment,

It may be a problem of current clang-format version, when formatting it moves the pointer there

pranavrth · 2024-11-14T10:28:31Z

src/rdkafka_mock_cgrp.c

+    rd_kafka_mock_cgrp_consumer_member_t *member,
+    rd_kafka_topic_partition_list_t *current_assignment,
+    int *member_epoch) {
+        rd_kafka_topic_partition_list_t *returned_assignment = NULL;


Better to name it assignments_to_return.

I can change it to assignment_to_return as assignment is a set, assignments are a set of sets, for example the assignments for all the group members, while assignment is for a single member.

pranavrth · 2024-11-14T10:38:24Z

src/rdkafka_mock_handlers.c

+                        } else {
+                                rd_kafka_mock_cgrp_consumer_member_leave(
+                                    mcgrp, member);
+                                member = NULL;
+                        }


In the static membership case (-2), the member can rejoin. We shouldn't destroy the member. You can add a FIXME for now here.

I've implemented it and added some tests for static group membership with the mock cluster

src/rdkafka_mock_cgrp.c

pranavrth · 2024-11-15T01:34:29Z

tests/0120-asymmetric_subscription.c

+        if (test_consumer_group_protocol_classic()) {
+                do_test_asymmetric("range", bootstraps);
+                do_test_asymmetric("roundrobin", bootstraps);
+                do_test_asymmetric("cooperative-sticky", bootstraps);
+        }


We are not checking anything with this test now for the new protocol. I think we should enable "cooperative-sticky" test for the new protocol.

Before it was the range assignor that was tested, but I removed it, because, as said, we don't need to test the mock implementation, unless it's during development for making client implementation debugging more easy. For example I added the unit tests for that at the beginning to make sure we tested with a mock implementation without flaws.

If you prefer I can keep the range test as before as the simple mock implementation is range based

pranavrth · 2024-11-15T02:05:15Z

tests/0147-consumer_group_consumer_mock.c

I have not reviewed this file. I wanted to review this separately. Can you add a new PR for this test?

Could you review it here? We need to merge it too and it's related to this PR

Moved it to #4920

pranavrth · 2024-11-18T07:14:37Z

.semaphore/semaphore.yml

+            - make -j -C tests build
+            - make -C tests run_local_quick
+            - DESTDIR="$PWD/dest" make install
+            - export TEST_KAFKA_VERSION=4.0.0


Where is this being used? 4.0.0 is not released yet.

This is used in tests to check if a test can run or not, if for example it's >= a specific version, given we're using master I used the next planned release here, but could have used 9.9.9 too

pranavrth · 2024-11-18T07:20:27Z

src/rdkafka.h

Do we really want to expose these functions? Are we exposing for the tests only?

No, in Java they're public too see ConsumerGroupMetadata

- Mock handler implementation - Rename current consumer protocol from generic to classic - Mock handler with automatic or manual assignment - More consumer group metadata getters - Test helpers - Expedite next HB after FindCoordinator doing it with an exponential backoff to avoid tight loops - Configurable session timeout and HB interval - Fix mock handler ListOffsets response LeaderEpoch instead of CurrentLeaderEpoch - Integration tests passing with AK trunk - Improve documentation and KIP 848 specific mock tests - Add mock tests for unknown topic id in metadata request and partial reconciliation - Make test 0147 more reliable - Fix test 0106 after HB timeout change - Exclude test case with AK trunk - Rename rd_kafka_buf_write_tags to rd_kafka_buf_write_tags_empty - Trivup 0.12.5 can run a KafkaCluster directly with KRaft and AK trunk - Trivup 0.12.6 build with a specific commit

and Py 3.12

an issue with apache/kafka#16464 on AK > 3.8.0

Address comment about using rd_kafka_buf_read_topic_partitions Asserts on non-nullable fields

…ava client

…ent_new

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from 6667956 to ca1eb2d Compare March 26, 2024 17:48

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch 2 times, most recently from 1d9962c to ca5e6d0 Compare March 27, 2024 10:43

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from ca1eb2d to 953f668 Compare March 28, 2024 14:41

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch 6 times, most recently from fd13d77 to f519c11 Compare March 28, 2024 22:35

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from 953f668 to e4614be Compare April 1, 2024 14:40

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch from f519c11 to 029b5ab Compare April 1, 2024 14:41

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from e4614be to 10e7cc1 Compare April 1, 2024 15:07

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch from 029b5ab to d7428d0 Compare April 1, 2024 15:07

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from 10e7cc1 to 5789307 Compare April 1, 2024 16:48

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch from d7428d0 to 0379129 Compare April 1, 2024 16:50

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch 2 times, most recently from 434da75 to e5f3101 Compare April 3, 2024 13:40

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch from 0379129 to c6c9d2e Compare April 3, 2024 13:41

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from e5f3101 to 109d348 Compare April 3, 2024 16:54

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch from c6c9d2e to cf1ad35 Compare April 3, 2024 16:54

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from 109d348 to 3dd0e45 Compare April 5, 2024 08:28

emasab changed the base branch from dev_kip848_use_metadata_cache_and_fixes to master April 5, 2024 08:38

emasab changed the base branch from master to dev_kip848_use_metadata_cache_and_fixes April 5, 2024 08:39

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from 3dd0e45 to 3bba91f Compare April 5, 2024 19:19

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch from 0396b1f to 27817d0 Compare April 5, 2024 19:21

emasab force-pushed the dev_kip848_use_metadata_cache_and_fixes branch from 3bba91f to f4fb7a1 Compare April 5, 2024 19:52

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch 3 times, most recently from 91df086 to 82774f9 Compare April 6, 2024 12:14

emasab requested a review from a team as a code owner July 1, 2024 10:13

emasab force-pushed the dev_kip848_mock_handler_and_integration_tests branch from 1d34b16 to 1f8a2a4 Compare July 1, 2024 10:29

pranavrth requested changes Jul 17, 2024

View reviewed changes

pranavrth requested changes Jul 18, 2024

View reviewed changes

pranavrth reviewed Jul 25, 2024

View reviewed changes

airlock-confluentinc bot force-pushed the dev_kip848_mock_handler_and_integration_tests branch from 1f8a2a4 to d72576a Compare August 5, 2024 14:42

airlock-confluentinc bot force-pushed the dev_kip848_mock_handler_and_integration_tests branch from aa99639 to 19ab07f Compare October 3, 2024 19:00

airlock-confluentinc bot force-pushed the dev_kip848_mock_handler_and_integration_tests branch from 19ab07f to 3d70ed0 Compare October 18, 2024 08:28

airlock-confluentinc bot force-pushed the dev_kip848_mock_handler_and_integration_tests branch from 53ea4ba to 50c4130 Compare October 28, 2024 20:24

emasab requested a review from pranavrth October 28, 2024 20:26

pranavrth reviewed Nov 14, 2024

View reviewed changes

pranavrth reviewed Nov 15, 2024

View reviewed changes

pranavrth requested changes Nov 20, 2024

View reviewed changes

emasab added 13 commits November 29, 2024 17:39

trivup 0.12.7 with fixes for AK 3.8.0

aa9e476

and Py 3.12

New version of trivup 0.12.7 to fix

4d922ee

an issue with apache/kafka#16464 on AK > 3.8.0

Address comments

a50d234

Static group membership mock tests

0c70015

Address comment about using rd_kafka_buf_read_topic_partitions Asserts on non-nullable fields

Leave only the rd_kafka_consumer_group_metadata_generation_id as in J…

548fc15

…ava client

Rename simple mock assignment

6d2e1c7

Rename mock configuration setters

ab0d6fb

Change parameters order in rd_kafka_mock_cgrp_consumer_target_assignm…

2e8bfb0

…ent_new

Rename to assignment_to_return

34e2c84

Test the range assignor with 848 mock cluster too

e5079aa

Variable naming

f59d7d4

Move test 0147 to a different PR

07f402b

airlock-confluentinc bot force-pushed the dev_kip848_mock_handler_and_integration_tests branch from a865822 to 07f402b Compare November 29, 2024 16:56

		rd_kafka_topic_partition_list_t *
		target_assignment; /**< Target assignment,

[KIP-848] Mock handler and Integration tests passing #4662

Are you sure you want to change the base?

[KIP-848] Mock handler and Integration tests passing #4662

Conversation

emasab commented Mar 26, 2024 • edited Loading

pranavrth left a 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

emasab Nov 21, 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

emasab Oct 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

emasab Nov 21, 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

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

pranavrth left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

emasab Oct 28, 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

emasab Nov 21, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pranavrth left a 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

confluent-cla-assistant bot commented Oct 3, 2024

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

emasab Nov 21, 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

emasab commented Mar 26, 2024 •

edited

Loading

emasab Nov 21, 2024 •

edited

Loading

emasab Oct 24, 2024 •

edited

Loading

emasab Nov 21, 2024 •

edited

Loading

emasab Oct 28, 2024 •

edited

Loading

emasab Nov 21, 2024 •

edited

Loading

emasab Nov 21, 2024 •

edited

Loading