You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2025/11/21 17:55:55 UTC
[PR] KAFKA-19903: Integration tests for share group throttled delivery [kafka]
apoorvmittal10 opened a new pull request, #20953:
URL: https://github.com/apache/kafka/pull/20953
Integartion tests for share group throttled delivery when delivery count is reaching the max delivery limit.
The changes were introduced in the PR: https://github.com/apache/kafka/pull/20837
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19903: Integration tests for share group throttled delivery [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20953:
URL: https://github.com/apache/kafka/pull/20953#discussion_r2553129841
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -3435,6 +3435,315 @@ public void testSharePartitionLagOnShareCoordinatorMovement() {
}
}
+ @ClusterTest
+ public void testFetchWithThrottledDelivery() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ // Produce a batch of 100 messages
+ for (int i = 0; i < 100; i++) {
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), ("Message " + i).getBytes());
+ producer.send(record);
+ }
+ producer.flush();
+
+ shareConsumer.subscribe(List.of(tp.topic()));
+ // Fetch records in 5 iterations, each time acknowledging with RELEASE. 5 is the default
+ // delivery limit hence we should see throttling from Math.ceil(5/2) = 3 fetches.
+ int throttleDeliveryLimit = 3;
+ for (int i = 0; i < 5; i++) {
+ // Adjust expected fetch count based on throttling. If i < throttleDeliveryLimit, we get full batch of 100.
+ // If i == 4 i.e. the last delivery, then we get 1 record.
+ // Otherwise, we get half the previous fetch count due to throttling. In this case, 100 >> (i - throttleDeliveryLimit + 1) it is 50 for i=3.
+ int expectedFetchCount = (i < throttleDeliveryLimit) ? 100 : ((i == 4) ? 1 : 50);
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, expectedFetchCount);
+ assertEquals(expectedFetchCount, records.count());
+
+ records.forEach(record -> shareConsumer.acknowledge(record, AcknowledgeType.RELEASE));
+ Map<TopicIdPartition, Optional<KafkaException>> result = shareConsumer.commitSync();
+ assertEquals(1, result.size());
+ assertEquals(Optional.empty(),
+ result.get(new TopicIdPartition(tpId, tp.partition(), tp.topic())));
+ }
+
+ // Offset 0 has already reached the delivery limit hence shall be archived.
+ // Offset 1 to 49 shall be in last deliver attempt and hence 1 record per poll.
+ validateExpectedRecordsInEachPollAndRelease(shareConsumer, 1, 50, 1);
+ // Delivery limit 4.
+ validateExpectedRecordsInEachPollAndRelease(shareConsumer, 50, 100, 50);
+ // Delivery limit 5.
+ validateExpectedRecordsInEachPollAndRelease(shareConsumer, 50, 100, 1);
+ // Next poll should not have any records as all records have reached delivery limit.
+ ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(2500L));
+ assertTrue(records.isEmpty(), "Records should be empty as all records have reached delivery limit. But received: " + records.count());
+ }
+ }
+
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key = "group.share.delivery.count.limit", value = "10"),
+ }
+ )
+ public void testFetchWithThrottledDeliveryBatchesWithIncreasedDeliveryLimit() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ // Produce records in complete power of 2 to fully test the throttling behavior.
+ int producedMessageCount = 512;
+ // Produce a batch of 512 messages
Review Comment:
I changed the max.poll.records to 512 for this test so it shouldn't fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19903: Integration tests for share group throttled delivery [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 merged PR #20953:
URL: https://github.com/apache/kafka/pull/20953
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19903: Integration tests for share group throttled delivery [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20953:
URL: https://github.com/apache/kafka/pull/20953#discussion_r2551080131
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -3435,6 +3435,315 @@ public void testSharePartitionLagOnShareCoordinatorMovement() {
}
}
+ @ClusterTest
+ public void testFetchWithThrottledDelivery() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ // Produce a batch of 100 messages
+ for (int i = 0; i < 100; i++) {
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), ("Message " + i).getBytes());
+ producer.send(record);
+ }
+ producer.flush();
+
+ shareConsumer.subscribe(List.of(tp.topic()));
+ // Fetch records in 5 iterations, each time acknowledging with RELEASE. 5 is the default
+ // delivery limit hence we should see throttling from Math.ceil(5/2) = 3 fetches.
+ int throttleDeliveryLimit = 3;
+ for (int i = 0; i < 5; i++) {
+ // Adjust expected fetch count based on throttling. If i < throttleDeliveryLimit, we get full batch of 100.
+ // If i == 4 i.e. the last delivery, then we get 1 record.
+ // Otherwise, we get half the previous fetch count due to throttling. In this case, 100 >> (i - throttleDeliveryLimit + 1) it is 50 for i=3.
+ int expectedFetchCount = (i < throttleDeliveryLimit) ? 100 : ((i == 4) ? 1 : 50);
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, expectedFetchCount);
+ assertEquals(expectedFetchCount, records.count());
+
+ records.forEach(record -> shareConsumer.acknowledge(record, AcknowledgeType.RELEASE));
+ Map<TopicIdPartition, Optional<KafkaException>> result = shareConsumer.commitSync();
+ assertEquals(1, result.size());
+ assertEquals(Optional.empty(),
+ result.get(new TopicIdPartition(tpId, tp.partition(), tp.topic())));
+ }
+
+ // Offset 0 has already reached the delivery limit hence shall be archived.
+ // Offset 1 to 49 shall be in last deliver attempt and hence 1 record per poll.
Review Comment:
nit: delivery attempt, not deliver attempt
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -3435,6 +3435,315 @@ public void testSharePartitionLagOnShareCoordinatorMovement() {
}
}
+ @ClusterTest
+ public void testFetchWithThrottledDelivery() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ // Produce a batch of 100 messages
+ for (int i = 0; i < 100; i++) {
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), ("Message " + i).getBytes());
+ producer.send(record);
+ }
+ producer.flush();
+
+ shareConsumer.subscribe(List.of(tp.topic()));
+ // Fetch records in 5 iterations, each time acknowledging with RELEASE. 5 is the default
+ // delivery limit hence we should see throttling from Math.ceil(5/2) = 3 fetches.
+ int throttleDeliveryLimit = 3;
+ for (int i = 0; i < 5; i++) {
+ // Adjust expected fetch count based on throttling. If i < throttleDeliveryLimit, we get full batch of 100.
+ // If i == 4 i.e. the last delivery, then we get 1 record.
+ // Otherwise, we get half the previous fetch count due to throttling. In this case, 100 >> (i - throttleDeliveryLimit + 1) it is 50 for i=3.
+ int expectedFetchCount = (i < throttleDeliveryLimit) ? 100 : ((i == 4) ? 1 : 50);
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, expectedFetchCount);
+ assertEquals(expectedFetchCount, records.count());
+
+ records.forEach(record -> shareConsumer.acknowledge(record, AcknowledgeType.RELEASE));
+ Map<TopicIdPartition, Optional<KafkaException>> result = shareConsumer.commitSync();
+ assertEquals(1, result.size());
+ assertEquals(Optional.empty(),
+ result.get(new TopicIdPartition(tpId, tp.partition(), tp.topic())));
+ }
+
+ // Offset 0 has already reached the delivery limit hence shall be archived.
+ // Offset 1 to 49 shall be in last deliver attempt and hence 1 record per poll.
+ validateExpectedRecordsInEachPollAndRelease(shareConsumer, 1, 50, 1);
+ // Delivery limit 4.
+ validateExpectedRecordsInEachPollAndRelease(shareConsumer, 50, 100, 50);
+ // Delivery limit 5.
+ validateExpectedRecordsInEachPollAndRelease(shareConsumer, 50, 100, 1);
+ // Next poll should not have any records as all records have reached delivery limit.
+ ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(Duration.ofMillis(2500L));
+ assertTrue(records.isEmpty(), "Records should be empty as all records have reached delivery limit. But received: " + records.count());
+ }
+ }
+
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key = "group.share.delivery.count.limit", value = "10"),
+ }
+ )
+ public void testFetchWithThrottledDeliveryBatchesWithIncreasedDeliveryLimit() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ // Produce records in complete power of 2 to fully test the throttling behavior.
+ int producedMessageCount = 512;
+ // Produce a batch of 512 messages
Review Comment:
I think it would be prudent to keep an eye out for flakiness here. I think we are depending on a single batch being written which will be true if the records are all sent within the linger delay.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19903: Integration tests for share group throttled delivery [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #20953:
URL: https://github.com/apache/kafka/pull/20953#discussion_r2553129637
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -3435,6 +3435,315 @@ public void testSharePartitionLagOnShareCoordinatorMovement() {
}
}
+ @ClusterTest
+ public void testFetchWithThrottledDelivery() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+ "group1",
+ Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT))
+ ) {
+ // Produce a batch of 100 messages
+ for (int i = 0; i < 100; i++) {
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), ("Message " + i).getBytes());
+ producer.send(record);
+ }
+ producer.flush();
+
+ shareConsumer.subscribe(List.of(tp.topic()));
+ // Fetch records in 5 iterations, each time acknowledging with RELEASE. 5 is the default
+ // delivery limit hence we should see throttling from Math.ceil(5/2) = 3 fetches.
+ int throttleDeliveryLimit = 3;
+ for (int i = 0; i < 5; i++) {
+ // Adjust expected fetch count based on throttling. If i < throttleDeliveryLimit, we get full batch of 100.
+ // If i == 4 i.e. the last delivery, then we get 1 record.
+ // Otherwise, we get half the previous fetch count due to throttling. In this case, 100 >> (i - throttleDeliveryLimit + 1) it is 50 for i=3.
+ int expectedFetchCount = (i < throttleDeliveryLimit) ? 100 : ((i == 4) ? 1 : 50);
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer, 2500L, expectedFetchCount);
+ assertEquals(expectedFetchCount, records.count());
+
+ records.forEach(record -> shareConsumer.acknowledge(record, AcknowledgeType.RELEASE));
+ Map<TopicIdPartition, Optional<KafkaException>> result = shareConsumer.commitSync();
+ assertEquals(1, result.size());
+ assertEquals(Optional.empty(),
+ result.get(new TopicIdPartition(tpId, tp.partition(), tp.topic())));
+ }
+
+ // Offset 0 has already reached the delivery limit hence shall be archived.
+ // Offset 1 to 49 shall be in last deliver attempt and hence 1 record per poll.
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19903: Integration tests for share group throttled delivery [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #20953:
URL: https://github.com/apache/kafka/pull/20953#issuecomment-3566695609
> I also would like an additional test.
>
> * Publish 2 records 0 and 1 in separate batches
> * Consumer1 (explicit mode, record limit, max poll = 1) fetches record 0 and does not yet ack
> * Consumer2 (explicit mode, record limit, max poll = 2) fetches record 1 and releases it repeatedly until the record is ready for its final delivery attempt
> * Consumer1 release its record, but does not poll yet
> * Consumer2 now again, but it gets record 0 only because record 1 is at its final delivery attempt and will no longer be combined with other records in its delivery, it accepts the record
> * Consumer2 now again fetch and it gets record 1 for its final delivery attempt and accepts, and closes
> * Consumer1 polls and finds no records
>
> wdyt?
Agree, it would be good test case. Added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org