KAFKA-19535: add integration tests for DescribeProducersOptions#brokerId by apalan60 · Pull Request #8 · apalan60/kafka

apalan60 · 2025-08-27T14:50:38Z

No description provided.

apalan60 · 2025-08-27T14:54:17Z

/review
auto_approve
--pr_reviewer.extra_instructions="
In the possible issues section, emphasize the following:

Is the code logic efficient?

"
--pr_reviewer.inline_code_comments=true
-i
--pr_reviewer.require_score_review=true
--pr_reviewer.require_can_be_split_review=true
--pr_reviewer.require_soc2_ticket=true
--pr_reviewer.num_code_suggestions="5"

qodo-code-review · 2025-08-27T14:54:22Z

Manual-approval option for PR-Agent is disabled. You can enable it via a configuration file

apalan60 · 2025-08-27T14:56:21Z

/gemini review

apalan60 · 2025-08-27T14:56:35Z

/improve
--pr_code_suggestions.extra_instructions="
Emphasize the following:

Does the code logic cover relevant edge cases?
Is the code logic clear and easy to understand?
Is the code logic efficient?

"
--pr_code_suggestions.num_code_suggestions_per_chunk="4"
--pr_code_suggestions.commitable_code_suggestions=true
--pr_code_suggestions.suggestions_score_threshold="0"

gemini-code-assist

Code Review

This pull request adds integration tests for the brokerId option in DescribeProducersOptions, which is a valuable addition. The tests cover the main scenarios, including routing to leaders, followers, and invalid brokers. I've identified a couple of areas where the tests can be improved for better type safety and more robust assertions. My feedback includes specific suggestions to address these points.

gemini-code-assist · 2025-08-27T14:57:33Z

+        this.clusterInstance = clusterInstance;
+    }
+
+    private static void sendTestRecords(Producer<Object, Object> producer) {


The producer is configured with StringSerializer for both key and value, so its type is Producer<String, String>. Using Producer<Object, Object> in the method signature is not type-safe. It's better to use the specific types to leverage Java's type system and prevent potential runtime errors.

Suggested change

private static void sendTestRecords(Producer<Object, Object> producer) {

private static void sendTestRecords(Producer<String, String> producer) {

gemini-code-assist · 2025-08-27T14:57:33Z

+            var followerProducerId = followerState.activeProducers().iterator().next().producerId();
+            var leaderProducerId = leaderState.activeProducers().iterator().next().producerId();
+            assertEquals(leaderProducerId, followerProducerId);


The current assertion only compares the producer ID of the first active producer. A more robust test would compare the entire list of active producers between the leader and the follower, similar to what is done in testDescribeProducersDefaultRoutesToLeader. This ensures that all producer states are correctly replicated to the follower.

Suggested change

var followerProducerId = followerState.activeProducers().iterator().next().producerId();

var leaderProducerId = leaderState.activeProducers().iterator().next().producerId();

assertEquals(leaderProducerId, followerProducerId);

assertEquals(leaderState.activeProducers(), followerState.activeProducers());

qodo-code-review · 2025-08-27T14:57:41Z

+            var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition))
+                    .partitionResult(topicPartition).get();


Suggestion: Await leader election and metadata stabilization before asserting producer state to avoid flakiness. Add a wait/retry until the partition has a known leader and at least one active producer is observed, with a reasonable timeout. [general, importance: 6]

Suggested change

var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition))

.partitionResult(topicPartition).get();

TestUtils.waitForCondition(() -> {

try {

var state = admin.describeProducers(List.of(topicPartition))

.partitionResult(topicPartition).get();

return state != null && !state.activeProducers().isEmpty();

} catch (Exception e) {

return false;

}

}, 10_000L, "Timed out waiting for active producers on leader");

var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition))

.partitionResult(topicPartition).get();

qodo-code-review · 2025-08-27T14:57:41Z

+            var followerState = admin.describeProducers(List.of(topicPartition), 
+                    new DescribeProducersOptions().brokerId(followerBrokerId))
+                    .partitionResult(topicPartition).get();
+            var leaderState = admin.describeProducers(List.of(topicPartition))
+                    .partitionResult(topicPartition).get();
+
+            assertNotNull(followerState);
+            assertFalse(followerState.activeProducers().isEmpty());
+
+            var followerProducerId = followerState.activeProducers().iterator().next().producerId();
+            var leaderProducerId = leaderState.activeProducers().iterator().next().producerId();
+            assertEquals(leaderProducerId, followerProducerId);


Suggestion: Guard against the case where leader has no active producers yet to prevent NoSuchElementException. Add waits ensuring both leader and follower responses have at least one active producer before comparing IDs. [general, importance: 7]

Suggested change

var followerState = admin.describeProducers(List.of(topicPartition),

new DescribeProducersOptions().brokerId(followerBrokerId))

.partitionResult(topicPartition).get();

var leaderState = admin.describeProducers(List.of(topicPartition))

.partitionResult(topicPartition).get();

assertNotNull(followerState);

assertFalse(followerState.activeProducers().isEmpty());

var followerProducerId = followerState.activeProducers().iterator().next().producerId();

var leaderProducerId = leaderState.activeProducers().iterator().next().producerId();

assertEquals(leaderProducerId, followerProducerId);

TestUtils.waitForCondition(() -> {

try {

var ls = admin.describeProducers(List.of(topicPartition)).partitionResult(topicPartition).get();

var fs = admin.describeProducers(List.of(topicPartition), new DescribeProducersOptions().brokerId(followerBrokerId))

.partitionResult(topicPartition).get();

return ls != null && fs != null && !ls.activeProducers().isEmpty() && !fs.activeProducers().isEmpty();

} catch (Exception e) {

return false;

}

}, 10_000L, "Timed out waiting for active producers on leader and follower");

var followerState = admin.describeProducers(List.of(topicPartition),

new DescribeProducersOptions().brokerId(followerBrokerId))

.partitionResult(topicPartition).get();

var leaderState = admin.describeProducers(List.of(topicPartition))

.partitionResult(topicPartition).get();

assertNotNull(followerState);

assertFalse(followerState.activeProducers().isEmpty());

assertFalse(leaderState.activeProducers().isEmpty());

var followerProducerId = followerState.activeProducers().iterator().next().producerId();

var leaderProducerId = leaderState.activeProducers().iterator().next().producerId();

assertEquals(leaderProducerId, followerProducerId);

qodo-code-review · 2025-08-27T14:57:41Z

+            var followerBrokerId = replicaBrokerIds.stream()
+                    .filter(id -> !id.equals(leaderBrokerId))
+                    .findFirst()
+                    .orElseThrow(() -> new IllegalStateException("No follower found"));


Suggestion: Ensure a true follower is selected when multiple replicas exist by excluding the leader and, if available, preferring an in-sync replica that is not leader. This avoids accidental selection of the leader after a leadership change. [general, importance: 1]

Suggested change

var followerBrokerId = replicaBrokerIds.stream()

.filter(id -> !id.equals(leaderBrokerId))

.findFirst()

.orElseThrow(() -> new IllegalStateException("No follower found"));

var followerBrokerId = replicaBrokerIds.stream()

.filter(id -> !id.equals(leaderBrokerId))

.findFirst()

.orElseThrow(() -> new IllegalStateException("No follower found for partition " + topicPartition));

qodo-code-review · 2025-08-27T14:57:42Z

+            TestUtils.assertFutureThrows(NotLeaderOrFollowerException.class, 
+                    admin.describeProducers(List.of(topicPartition), 
+                            new DescribeProducersOptions().brokerId(nonReplicaBrokerId))
+                            .partitionResult(topicPartition)); 


Suggestion: Wrap the future supplier to retry until metadata reflects the chosen non-replica to avoid transient leader movements causing different exceptions. Add a bounded wait ensuring the broker remains a non-replica before asserting the failure. [general, importance: 6]

Suggested change

TestUtils.assertFutureThrows(NotLeaderOrFollowerException.class,

admin.describeProducers(List.of(topicPartition),

new DescribeProducersOptions().brokerId(nonReplicaBrokerId))

.partitionResult(topicPartition));

// Ensure the selected broker is not a replica before asserting

TestUtils.waitForCondition(() -> {

try {

var desc = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME);

var currentReplicaIds = desc.partitions().get(0).replicas().stream().map(Node::id).toList();

return !currentReplicaIds.contains(nonReplicaBrokerId);

} catch (Exception e) {

return false;

}

}, 10_000L, "Timed out ensuring broker is non-replica");

TestUtils.assertFutureThrows(NotLeaderOrFollowerException.class,

admin.describeProducers(List.of(topicPartition),

new DescribeProducersOptions().brokerId(nonReplicaBrokerId))

.partitionResult(topicPartition));

apalan60 · 2025-08-27T14:58:53Z

/review
--pr_reviewer.extra_instructions="
In the possible issues section, emphasize the following:

Is the code logic efficient?

"
--pr_reviewer.inline_code_comments=true
-i
--pr_reviewer.require_score_review=true
--pr_reviewer.require_can_be_split_review=true
--pr_reviewer.require_soc2_ticket=false
--pr_reviewer.num_code_suggestions="5"

qodo-code-review · 2025-08-27T14:59:11Z

PR Reviewer Guide 🔍

(Review updated until commit `956c3aa`)

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 2 🔵🔵⚪⚪⚪
🧪 PR contains tests
🔒 No security concerns identified
🔀 No multiple PR themes
⚡ Recommended focus areas for review Efficiency The tests repeatedly describe topics and materialize full topic descriptions to derive replica IDs; consider caching the partition description or querying only once per test to reduce admin calls and speed up integration tests. var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME); var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream() .map(Node::id) .toList(); var leaderBrokerId = clusterInstance.getLeaderBrokerId(topicPartition); var followerBrokerId = replicaBrokerIds.stream() .filter(id -> id != leaderBrokerId) .findFirst() .orElseThrow(() -> new IllegalStateException("No follower found for partition " + topicPartition)); var followerState = admin.describeProducers(List.of(topicPartition), new DescribeProducersOptions().brokerId(followerBrokerId)) .partitionResult(topicPartition).get(); var leaderState = admin.describeProducers(List.of(topicPartition)) .partitionResult(topicPartition).get(); assertNotNull(followerState); assertFalse(followerState.activeProducers().isEmpty()); assertEquals(leaderState.activeProducers(), followerState.activeProducers()); } } @ClusterTest void testDescribeProducersWithInvalidBrokerId() throws Exception { clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR); try (Producer<byte[], byte[]> producer = clusterInstance.producer(); var admin = clusterInstance.admin()) { sendTestRecords(producer); var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME); var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream() .map(Node::id) .toList(); var nonReplicaBrokerId = clusterInstance.brokerIds().stream() .filter(id -> !replicaBrokerIds.contains(id)) .findFirst() .orElseThrow(() -> new IllegalStateException("No non-replica broker found")); Flakiness After sending records, the tests immediately call describeProducers; consider waiting for metadata/leader election or using TestUtils utilities to ensure partition leaders and replication are ready to reduce timing-related flakes. try (Producer<byte[], byte[]> producer = clusterInstance.producer(); var admin = clusterInstance.admin()) { sendTestRecords(producer); var stateWithExplicitLeader = admin.describeProducers(List.of(topicPartition), new DescribeProducersOptions().brokerId(clusterInstance.getLeaderBrokerId(topicPartition))) .partitionResult(topicPartition).get(); var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition)) .partitionResult(topicPartition).get(); assertNotNull(stateWithDefaultRouting); assertFalse(stateWithDefaultRouting.activeProducers().isEmpty()); assertEquals(stateWithExplicitLeader.activeProducers(), stateWithDefaultRouting.activeProducers()); }

apalan60 · 2025-08-27T15:25:30Z

/gemini review

gemini-code-assist

Code Review

This pull request adds integration tests for the brokerId option in DescribeProducersOptions. The tests cover scenarios for routing to the leader, a follower, and an invalid broker that is not a replica. The test logic is sound and covers the intended functionality well. I've provided a couple of suggestions to improve the test structure by reducing code duplication, which will enhance maintainability.

gemini-code-assist · 2025-08-27T15:27:14Z

+             var admin = clusterInstance.admin()) {
+            sendTestRecords(producer);
+
+            var topicPartition = new TopicPartition(TOPIC_NAME, 0);
+            var leaderBrokerId = clusterInstance.getLeaderBrokerId(topicPartition);
+


To reduce code duplication across the test methods, consider creating the topic, producer, and admin client in a @BeforeEach setup method and closing the clients in an @AfterEach teardown method. This would remove the need for clusterInstance.createTopic(...) and the try-with-resources block in each test, making the tests cleaner and more maintainable.

For example:

private Admin admin; private Producer<Object, Object> producer; @BeforeEach void setup() throws Exception { admin = clusterInstance.admin(); producer = clusterInstance.producer(Map.of( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class, ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class)); clusterInstance.createTopic(TOPIC_NAME, NUM_PARTITIONS, REPLICATION_FACTOR); } @AfterEach void tearDown() { if (producer != null) { producer.close(); } if (admin != null) { admin.close(); } }

gemini-code-assist · 2025-08-27T15:27:14Z

+                    .findFirst()
+                    .orElseThrow(() -> new IllegalStateException("No follower found for partition " + topicPartition));
+
+            var followerState = admin.describeProducers(List.of(topicPartition), 
+                    new DescribeProducersOptions().brokerId(followerBrokerId))


This block of code to get replica broker IDs is duplicated in testDescribeProducersWithInvalidBrokerId. To improve maintainability and reduce duplication, consider extracting this logic into a private helper method. Additionally, new TopicPartition(TOPIC_NAME, 0) is used in all tests and could be defined as a static final constant for the class.

For example:

private static final TopicPartition TOPIC_PARTITION_0 = new TopicPartition(TOPIC_NAME, 0); private List<Integer> getReplicaBrokerIds(Admin admin, TopicPartition topicPartition) throws Exception { TopicDescription topicDescription = admin.describeTopics(List.of(topicPartition.topic())) .allTopicNames().get().get(topicPartition.topic()); return topicDescription.partitions().get(topicPartition.partition()).replicas().stream() .map(Node::id) .toList(); }

Then you could simplify this block to:

var replicaBrokerIds = getReplicaBrokerIds(admin, TOPIC_PARTITION_0);

apalan60 · 2025-08-27T15:59:39Z

/gemini review

apalan60 · 2025-08-27T15:59:56Z

/review
--pr_reviewer.extra_instructions="
In the possible issues section, emphasize the following:

Is the code logic efficient?

"
--pr_reviewer.inline_code_comments=true
-i
--pr_reviewer.require_score_review=false
--pr_reviewer.require_can_be_split_review=true
--pr_reviewer.num_code_suggestions="5"

qodo-code-review · 2025-08-27T16:00:20Z

Persistent review updated to latest commit 956c3aa

gemini-code-assist

Code Review

This pull request adds integration tests for DescribeProducersOptions#brokerId. The tests cover the main scenarios: default routing to leader, querying a follower, and querying an invalid broker. The test logic is sound. I've provided a few suggestions to improve code quality by using explicit charsets, removing hardcoded values, and reducing code duplication. These changes will make the new test class more robust and maintainable.

gemini-code-assist · 2025-08-27T16:01:10Z

+
+    private static void sendTestRecords(Producer<byte[], byte[]> producer) {
+        for (int partition = 0; partition < NUM_PARTITIONS; partition++) {
+            producer.send(new ProducerRecord<>(TOPIC_NAME, partition, ("key-" + partition).getBytes(), ("value-" + partition).getBytes()));


Using String.getBytes() without specifying a charset can lead to platform-dependent behavior, as it uses the system's default charset. It's a best practice to always specify a charset, like StandardCharsets.UTF_8, to ensure consistent behavior across different environments.

Suggested change

producer.send(new ProducerRecord<>(TOPIC_NAME, partition, ("key-" + partition).getBytes(), ("value-" + partition).getBytes()));

producer.send(new ProducerRecord<>(TOPIC_NAME, partition, ("key-" + partition).getBytes(java.nio.charset.StandardCharsets.UTF_8), ("value-" + partition).getBytes(java.nio.charset.StandardCharsets.UTF_8)));

gemini-code-assist · 2025-08-27T16:01:10Z

+            var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME);
+            var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream()
+                    .map(Node::id)
+                    .toList();


This code block uses hardcoded values for the topic name (TOPIC_NAME) and partition index (0). It's better to use the topicPartition field to make the code more robust and readable.

Additionally, this logic is duplicated in testDescribeProducersWithInvalidBrokerId. Consider extracting it into a private helper method to improve maintainability.

Suggested change

var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME);

var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream()

.map(Node::id)

.toList();

var topicDescription = admin.describeTopics(List.of(topicPartition.topic())).allTopicNames().get().get(topicPartition.topic());

var replicaBrokerIds = topicDescription.partitions().get(topicPartition.partition()).replicas().stream()

.map(Node::id)

.toList();

gemini-code-assist · 2025-08-27T16:01:10Z

+            var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME);
+            var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream()
+                    .map(Node::id)
+                    .toList();


This code block uses hardcoded values for the topic name (TOPIC_NAME) and partition index (0). It's better to use the topicPartition field to make the code more robust and readable.

This logic is also duplicated from testDescribeProducersFromFollower. Consider extracting it into a private helper method to improve maintainability.

Suggested change

var topicDescription = admin.describeTopics(List.of(TOPIC_NAME)).allTopicNames().get().get(TOPIC_NAME);

var replicaBrokerIds = topicDescription.partitions().get(0).replicas().stream()

.map(Node::id)

.toList();

var topicDescription = admin.describeTopics(List.of(topicPartition.topic())).allTopicNames().get().get(topicPartition.topic());

var replicaBrokerIds = topicDescription.partitions().get(topicPartition.partition()).replicas().stream()

.map(Node::id)

.toList();

apalan60 · 2025-08-27T16:01:22Z

/describe
--pr_description.publish_labels=true
--pr_description.publish_description_as_comment=true
--pr_description.generate_ai_title=true
--pr_description.final_update_message=true

qodo-code-review · 2025-08-27T16:01:41Z

Title

Add integration tests for DescribeProducersOptions#brokerId

PR Type

Tests

Description

Add integration tests for DescribeProducersOptions#brokerId functionality
Test default routing to leader broker behavior
Test querying producer state from follower brokers
Test error handling for invalid broker IDs

Diagram Walkthrough

flowchart LR
  A["Test Setup"] --> B["Create Topic with Replicas"]
  B --> C["Send Test Records"]
  C --> D["Test Default Routing"]
  C --> E["Test Follower Query"]
  C --> F["Test Invalid Broker ID"]
  D --> G["Verify Leader Response"]
  E --> H["Verify Follower Response"]
  F --> I["Verify Exception Thrown"]

File Walkthrough

Relevant files

Tests

DescribeProducersWithBrokerIdTest.java `Integration tests for DescribeProducersOptions brokerId parameter` clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/DescribeProducersWithBrokerIdTest.java New test class with 3 test methods for `DescribeProducersOptions#brokerId` Tests default routing behavior to leader broker Tests querying producer state from follower brokers Tests error handling when using invalid broker IDs	+144/-0

apalan60 · 2025-08-27T16:21:40Z

./gemini review

gemini-code-assist · 2025-08-27T16:21:58Z

Hi @apalan60, I noticed you've re-triggered the /gemini review command. A review has already been provided and updated in previous comments (see comments #6 and #10).

Do you have any specific questions about the existing review, or have you pushed new changes that you'd like me to review again?

apalan60 · 2025-08-27T16:33:01Z

@gemini-code-assist

Hi @apalan60, I noticed you've re-triggered the /gemini review command. A review has already been provided and updated in previous comments (see comments #6 and apache#10).

Do you have any specific questions about the existing review, or have you pushed new changes that you'd like me to review again?

I made some update, hope to continous reivew

gemini-code-assist

Code Review

This pull request adds integration tests for the brokerId option in DescribeProducersOptions. The tests cover routing to the leader by default, explicitly routing to a follower, and handling invalid broker IDs. The test logic is sound and covers the main scenarios. I've provided a couple of suggestions to improve code maintainability by reducing duplication and to make the test setup more efficient.

gemini-code-assist · 2025-08-27T16:34:33Z

+        for (int partition = 0; partition < NUM_PARTITIONS; partition++) {
+            producer.send(new ProducerRecord<>(TOPIC_NAME, partition, ("key-" + partition).getBytes(), ("value-" + partition).getBytes()));
+        }


The tests in this class only use partition 0 (as defined by the topicPartition field). This loop sends records to all partitions, which is unnecessary for the test's purpose. To make the test setup more efficient, you can send a single record to partition 0.

producer.send(new ProducerRecord<>(TOPIC_NAME, 0, "key-0".getBytes(), "value-0".getBytes()));

gemini-code-assist · 2025-08-27T16:34:33Z

+            var topicDescription = admin.describeTopics(List.of(topicPartition.topic())).allTopicNames().get().get(topicPartition.topic());
+            var replicaBrokerIds = topicDescription.partitions().get(topicPartition.partition()).replicas().stream()
+                    .map(Node::id)
+                    .toList();


This block of code to retrieve replica broker IDs is duplicated from testDescribeProducersFromFollower (lines 100-103). To improve maintainability and reduce code duplication, consider extracting this logic into a private helper method.

For example:

private List<Integer> getReplicaBrokerIds(Admin admin) throws Exception { var topicDescription = admin.describeTopics(List.of(topicPartition.topic())).allTopicNames().get().get(topicPartition.topic()); return topicDescription.partitions().get(topicPartition.partition()).replicas().stream() .map(Node::id) .toList(); }

You can then call this method in both testDescribeProducersFromFollower and testDescribeProducersWithInvalidBrokerId.

apalan60 · 2025-09-09T13:28:16Z

-    private List<Integer> getReplicaBrokerIds(Admin admin, TopicPartition topicPartition) throws Exception {
-        var topicDescription = admin.describeTopics(List.of(topicPartition.topic())).allTopicNames().get().get(topicPartition.topic());
-        return topicDescription.partitions().get(topicPartition.partition()).replicas().stream()
+    private List<Integer> getReplicaBrokerIds(Admin admin) throws Exception {


test review

apalan60

test3

apalan60 · 2025-09-09T13:29:28Z

+        producer.flush();
+    }
+
+    @BeforeEach


apalan60 · 2025-09-09T13:29:57Z

+@ClusterTestDefaults(
+    brokers = 3
+)
+class DescribeProducersWithBrokerIdTest {


gemini-code-assist Bot reviewed Aug 27, 2025

View reviewed changes

qodo-code-review Bot reviewed Aug 27, 2025

View reviewed changes

qodo-code-review Bot added the Review effort 2/5 label Aug 27, 2025

apalan60 force-pushed the KAFKA-19535 branch from 2dd1719 to 88ac5b7 Compare August 27, 2025 15:26

gemini-code-assist Bot reviewed Aug 27, 2025

View reviewed changes

apalan60 force-pushed the KAFKA-19535 branch from 88ac5b7 to 956c3aa Compare August 27, 2025 15:58

gemini-code-assist Bot reviewed Aug 27, 2025

View reviewed changes

qodo-code-review Bot added the Tests label Aug 27, 2025

apalan60 force-pushed the KAFKA-19535 branch from 956c3aa to 0209ec1 Compare August 27, 2025 16:21

gemini-code-assist Bot reviewed Aug 27, 2025

View reviewed changes

apalan60 force-pushed the KAFKA-19535 branch from 0209ec1 to 75caf69 Compare August 27, 2025 16:39

KAFKA-19535: add integration tests for DescribeProducersOptions#brokerId

2a119a7

apalan60 force-pushed the KAFKA-19535 branch from 75caf69 to 2a119a7 Compare August 27, 2025 16:41

apalan60 added 2 commits August 30, 2025 14:18

Address comments

03708d3

refactor: extract helper method for brokerID retrieval

107c9d5

apalan60 added 3 commits August 30, 2025 15:06

Merge remote-tracking branch 'upstream/trunk' into KAFKA-19535

e4cadae

Address comments

5d7ccd4

Merge remote-tracking branch 'upstream/trunk' into KAFKA-19535

b89cc95

apalan60 force-pushed the trunk branch from 8a7d0ac to 2ba30cc Compare September 2, 2025 09:40

apalan60 commented Sep 9, 2025

View reviewed changes

	private static void sendTestRecords(Producer<Object, Object> producer) {
	private static void sendTestRecords(Producer<String, String> producer) {

		var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition))
		.partitionResult(topicPartition).get();

-            var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition))
-                    .partitionResult(topicPartition).get();
+            TestUtils.waitForCondition(() -> {
+                try {
+                    var state = admin.describeProducers(List.of(topicPartition))
+                            .partitionResult(topicPartition).get();
+                    return state != null && !state.activeProducers().isEmpty();
+                } catch (Exception e) {
+                    return false;
+                }
+            }, 10_000L, "Timed out waiting for active producers on leader");
+            var stateWithDefaultRouting = admin.describeProducers(List.of(topicPartition))
+                    .partitionResult(topicPartition).get();

	producer.send(new ProducerRecord<>(TOPIC_NAME, partition, ("key-" + partition).getBytes(), ("value-" + partition).getBytes()));
	producer.send(new ProducerRecord<>(TOPIC_NAME, partition, ("key-" + partition).getBytes(java.nio.charset.StandardCharsets.UTF_8), ("value-" + partition).getBytes(java.nio.charset.StandardCharsets.UTF_8)));

Conversation

apalan60 commented Aug 27, 2025

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

qodo-code-review Bot commented Aug 27, 2025

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

qodo-code-review Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

qodo-code-review Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

qodo-code-review Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

qodo-code-review Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

qodo-code-review Bot commented Aug 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Reviewer Guide 🔍

(Review updated until commit 956c3aa)

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

qodo-code-review Bot commented Aug 27, 2025

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 27, 2025

Choose a reason for hiding this comment

Uh oh!

apalan60 commented Aug 27, 2025

Uh oh!

qodo-code-review Bot commented Aug 27, 2025

Title

PR Type

Description

Diagram Walkthrough

File Walkthrough

Uh oh!

qodo-code-review Bot commented Aug 27, 2025 •

edited

Loading

(Review updated until commit `956c3aa`)