You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cychiu8 (via GitHub)" <gi...@apache.org> on 2026/04/15 01:27:09 UTC
[PR] [Test] the upload of gradle report when jdk version is different between trunk and 40 [kafka]
cychiu8 opened a new pull request, #22062:
URL: https://github.com/apache/kafka/pull/22062
Summary: test whether the gradle report is uploaded when jdk version is different between trunk and 4.0
--
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] [Test] the upload of gradle report when jdk version is different between trunk and 40 [kafka]
Posted by "cychiu8 (via GitHub)" <gi...@apache.org>.
cychiu8 closed pull request #22062: [Test] the upload of gradle report when jdk version is different between trunk and 40
URL: https://github.com/apache/kafka/pull/22062
--
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] [Test] the upload of gradle report when jdk version is different between trunk and 40 [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #22062:
URL: https://github.com/apache/kafka/pull/22062#discussion_r3083426500
##########
.github/workflows/deflake.yml:
##########
@@ -60,26 +65,52 @@ jobs:
TEST_PATTERN: ${{ inputs.test-pattern }}
run: |
set +e
- ./gradlew --info --build-cache --scan --continue \
+ ./.github/scripts/thread-dump.sh &
+ timeout 60m ./gradlew --info --build-cache --scan --continue \
-PtestLoggingEvents=started,passed,skipped,failed \
- -PignoreFailures=true -PmaxParallelForks=2 \
+ -PmaxParallelForks=2 \
-Pkafka.cluster.test.repeat=$TEST_REPEAT \
-PmaxTestRetries=$TEST_REPEAT -PmaxTestRetryFailures=0 \
- -PmaxQuarantineTestRetries=$TEST_REPEAT -PmaxQuarantineTestRetryFailures=0 \
- ${TEST_MODULE}:test ${TEST_MODULE}:quarantinedTest --tests $TEST_PATTERN
+ ${TEST_MODULE}:test --tests $TEST_PATTERN
exitcode="$?"
echo "exitcode=$exitcode" >> $GITHUB_OUTPUT
- - name: Archive JUnit reports
+
+ - name: Archive JUnit HTML reports
+ uses: actions/upload-artifact@v4
+ id: archive-junit-html
+ with:
+ name: junit-html-reports
+ path: |
+ **/build/reports/tests/*
+ compression-level: 9
+ if-no-files-found: ignore
+
+ - name: Archive JUnit XML
+ uses: actions/upload-artifact@v4
+ with:
+ name: junit-xml
+ path: |
+ build/junit-xml/**/*.xml
+ compression-level: 9
+ if-no-files-found: ignore
+
+ - name: Archive Thread Dumps
+ id: archive-thread-dump
+ if: steps.junit-test.outputs.gradle-exitcode == '124'
uses: actions/upload-artifact@v4
- id: junit-upload-artifact
with:
- name: junit-reports-${{ inputs.java-version }}
+ name: junit-thread-dumps-${{ env.job-variation }}
path: |
- **/build/reports/tests/test/*
+ thread-dumps/*
+ compression-level: 9
if-no-files-found: ignore
+
- name: Parse JUnit tests
- run: python .github/scripts/junit.py >> $GITHUB_STEP_SUMMARY
env:
GITHUB_WORKSPACE: ${{ github.workspace }}
- REPORT_URL: ${{ steps.junit-upload-artifact.outputs.artifact-url }}
- GRADLE_EXIT_CODE: ${{ steps.junit-test.outputs.exitcode }}
+ JUNIT_REPORT_URL: ${{ steps.archive-junit-html.outputs.artifact-url }}
+ THREAD_DUMP_URL: ${{ steps.archive-thread-dump.outputs.artifact-url }}
+ GRADLE_TEST_EXIT_CODE: ${{ steps.junit-test.outputs.gradle-exitcode }}
Review Comment:
The `junit-test` step writes `exitcode` to `$GITHUB_OUTPUT`, so the output key is `exitcode`, not `gradle-exitcode`. As written, the timeout detection and JUnit parsing will treat the exit code as missing and may fail the workflow incorrectly. Update these references to `steps.junit-test.outputs.exitcode` (or rename the output variable to match).
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/CreateTopicsRequestWithPolicyTest.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.CreateTopicsOptions;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.InvalidPartitionsException;
+import org.apache.kafka.common.errors.InvalidReplicationFactorException;
+import org.apache.kafka.common.errors.PolicyViolationException;
+import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.policy.CreateTopicPolicy;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ClusterTestDefaults(
+ brokers = 3,
+ serverProperties = {
+ @ClusterConfigProperty(key = ServerLogConfigs.CREATE_TOPIC_POLICY_CLASS_NAME_CONFIG, value = "org.apache.kafka.clients.CreateTopicsRequestWithPolicyTest$Policy"),
+ }
+)
+public class CreateTopicsRequestWithPolicyTest {
+
+ public static class Policy implements CreateTopicPolicy {
+ private Map<String, ?> configs;
+ private boolean closed = false;
+
+ @Override
+ public void configure(Map<String, ?> configs) {
+ this.configs = configs;
+ }
+
+ @Override
+ public void validate(RequestMetadata requestMetadata) throws PolicyViolationException {
+ if (Topic.isInternal(requestMetadata.topic())) {
+ // Do not verify internal topics
+ return;
+ }
+
+ if (closed) {
+ throw new IllegalStateException("Policy should not be closed");
+ }
+ if (configs == null || configs.isEmpty()) {
+ throw new IllegalStateException("Configure should have been called with non empty configs");
+ }
+
+ if (requestMetadata.numPartitions() != null || requestMetadata.replicationFactor() != null) {
+ assertNotNull(requestMetadata.numPartitions(), "numPartitions should not be null, but it is " + requestMetadata.numPartitions());
+ assertNotNull(requestMetadata.replicationFactor(), "replicationFactor should not be null, but it is " + requestMetadata.replicationFactor());
+ assertNull(requestMetadata.replicasAssignments(), "replicaAssignments should be null, but it is " + requestMetadata.replicasAssignments());
+
+ if (requestMetadata.numPartitions() < 5) {
+ throw new PolicyViolationException("Topics should have at least 5 partitions, received " +
+ requestMetadata.numPartitions());
+ }
+
+ if (requestMetadata.numPartitions() > 10) {
+ String retentionMs = requestMetadata.configs().get(TopicConfig.RETENTION_MS_CONFIG);
+ if (retentionMs == null || Integer.parseInt(retentionMs) > 5000) {
+ throw new PolicyViolationException("RetentionMs should be less than 5000ms if partitions > 10");
+ }
+ } else {
+ assertTrue(requestMetadata.configs().isEmpty(), "Topic configs should be empty, but it is " + requestMetadata.configs());
+ }
+ } else {
+ assertNull(requestMetadata.numPartitions(), "numPartitions should be null, but it is " + requestMetadata.numPartitions());
+ assertNull(requestMetadata.replicationFactor(), "replicationFactor should be null, but it is " + requestMetadata.replicationFactor());
+ assertNotNull(requestMetadata.replicasAssignments(), "replicasAssignments should not be null, but it is " + requestMetadata.replicasAssignments());
+
+ requestMetadata.replicasAssignments().forEach((partitionId, replicas) -> {
+ if (replicas.size() < 2) {
+ throw new PolicyViolationException("Topic partitions should have at least 2 replicas, received " +
+ replicas.size() + " for partition " + partitionId);
+ }
+ });
+ }
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+
+ private void validateValidCreateTopicsRequests(NewTopic topic, Admin admin, boolean validateOnly) throws Exception {
+ admin.createTopics(
+ List.of(topic),
+ new CreateTopicsOptions().validateOnly(validateOnly)
+ ).all().get();
+ }
+
+ @ClusterTest
+ public void testValidCreateTopicsRequests(ClusterInstance cluster) throws Exception {
+ try (Admin admin = cluster.admin()) {
+ cluster.createTopic("topic1", 5, (short) 1);
+
+ validateValidCreateTopicsRequests(
+ new NewTopic("topic2", 5, (short) 3),
+ admin,
+ true
+ );
+
+ validateValidCreateTopicsRequests(
+ new NewTopic("topic3", 11, (short) 2)
+ .configs(Map.of(TopicConfig.RETENTION_MS_CONFIG, "4999")),
+ admin,
+ true
+ );
+
+ validateValidCreateTopicsRequests(
+ new NewTopic("topic4", Map.of(
+ 0, List.of(1, 0),
+ 1, List.of(0, 1)
+ )),
+ admin,
+ false
+ );
+ }
+ }
+
+ private void validateErrorCreateTopicsRequests(NewTopic topic, Admin admin, boolean validateOnly, Class<? extends Throwable> expectedExceptionClass, String expectedErrorMessage) {
+ ExecutionException exception = assertThrows(ExecutionException.class, () ->
+ admin.createTopics(List.of(topic), new CreateTopicsOptions().validateOnly(validateOnly)).all().get());
+ assertEquals(
+ expectedExceptionClass,
+ exception.getCause().getClass(),
+ "Expected " + expectedExceptionClass.getSimpleName() + ", but got " + exception.getCause().getClass().getSimpleName()
+ );
+ assertTrue(exception.getMessage().contains(expectedErrorMessage));
Review Comment:
`ExecutionException.getMessage()` often includes wrapper text and may not reliably contain the underlying broker error message; the more stable check is against `exception.getCause().getMessage()`. Using the wrapper message risks brittle failures across JDKs/implementations.
```suggestion
assertNotNull(exception.getCause());
assertEquals(
expectedExceptionClass,
exception.getCause().getClass(),
"Expected " + expectedExceptionClass.getSimpleName() + ", but got " + exception.getCause().getClass().getSimpleName()
);
assertTrue(exception.getCause().getMessage().contains(expectedErrorMessage));
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/RackAwareAutoTopicCreationTest.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.server.config.ReplicationConfigs;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.server.config.ServerLogConfigs;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@ClusterTestDefaults(types = {Type.KRAFT},
+ brokers = 4,
+ serverProperties = {
+ @ClusterConfigProperty(key = ServerLogConfigs.NUM_PARTITIONS_CONFIG, value = "8"),
+ @ClusterConfigProperty(key = ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, value = "2"),
+ @ClusterConfigProperty(id = 0, key = ServerConfigs.BROKER_RACK_CONFIG, value = "0"),
+ @ClusterConfigProperty(id = 1, key = ServerConfigs.BROKER_RACK_CONFIG, value = "0"),
+ @ClusterConfigProperty(id = 2, key = ServerConfigs.BROKER_RACK_CONFIG, value = "1"),
+ @ClusterConfigProperty(id = 3, key = ServerConfigs.BROKER_RACK_CONFIG, value = "1"),
+ })
+public class RackAwareAutoTopicCreationTest {
+
+ private static final String TOPIC = "topic";
+
+ @ClusterTest
+ public void testAutoCreateTopic(ClusterInstance cluster) throws Exception {
+
+ try (Admin admin = cluster.admin();
+ Producer<byte[], byte[]> producer = cluster.producer()) {
+
+ // send a record to trigger auto create topic
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(TOPIC, null, "key".getBytes(), "value".getBytes());
+ assertEquals(0L, producer.send(record).get().offset(), "Should have offset 0");
+
+ // check broker rack content
+ Map<Integer, String> expectedBrokerToRackMap = Map.of(
+ 0, "0",
+ 1, "0",
+ 2, "1",
+ 3, "1"
+ );
+ Map<Integer, String> actualBrokerToRackMap = getBrokerToRackMap(cluster);
+ assertEquals(expectedBrokerToRackMap, actualBrokerToRackMap);
+
+ // get topic assignments and check it's content
+ Map<Integer, List<Integer>> assignments = getTopicAssignment(admin);
+ for (List<Integer> brokerList : assignments.values()) {
+ assertEquals(new HashSet<>(brokerList).size(), brokerList.size(),
+ "More than one replica is assigned to same broker for the same partition");
+ }
+
+ // check rack count for each partition
+ ReplicaDistributions distribution = getReplicaDistribution(assignments, expectedBrokerToRackMap);
+ Map<String, String> serverProperties = cluster.config().serverProperties();
+ int numPartition = Integer.parseInt(serverProperties.get(ServerLogConfigs.NUM_PARTITIONS_CONFIG));
+ int replicationFactor = Integer.parseInt(serverProperties.get(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG));
+ List<Integer> expectedRackCounts = Collections.nCopies(numPartition, replicationFactor);
+ List<Integer> actualRackCounts = distribution.partitionToRackMap.values().stream()
+ .map(racks -> (int) racks.stream().distinct().count())
+ .collect(Collectors.toList());
+ assertEquals(expectedRackCounts, actualRackCounts,
+ "More than one replica of the same partition is assigned to the same rack");
+
+ // check replica count for each partition
+ int numBrokers = cluster.brokers().size();
+ int numReplicasPerBroker = numPartition * replicationFactor / numBrokers;
+ List<Integer> expectedReplicasCounts = Collections.nCopies(numBrokers, numReplicasPerBroker);
+ List<Integer> actualReplicasCounts = new ArrayList<>(distribution.partitionToCountMap.values());
Review Comment:
`partitionToCountMap` is a `HashMap`, so `values()` iteration order is nondeterministic. Comparing it to an ordered `expectedReplicasCounts` list will be flaky/incorrect. Build `actualReplicasCounts` in a deterministic broker-id order (e.g., loop brokerIds sorted and `getOrDefault`) or compare as maps instead of lists.
```suggestion
List<Integer> actualReplicasCounts = actualBrokerToRackMap.keySet().stream()
.sorted()
.map(brokerId -> distribution.partitionToCountMap.getOrDefault(brokerId, 0))
.collect(Collectors.toList());
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/TransactionsWithMaxInFlightOneTest.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
+import org.apache.kafka.coordinator.transaction.TransactionStateManagerConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.test.TestUtils;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ClusterTestDefaults(
+ types = {Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "false"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = "log.unclean.leader.election.enable", value = "false"),
+ @ClusterConfigProperty(key = ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG, value = "false"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "0"),
+ @ClusterConfigProperty(key = TransactionStateManagerConfig.TRANSACTIONS_ABORT_TIMED_OUT_TRANSACTION_CLEANUP_INTERVAL_MS_CONFIG, value = "200")
+ }
+)
+public class TransactionsWithMaxInFlightOneTest {
+ private static final String TOPIC1 = "topic1";
+ private static final String TOPIC2 = "topic2";
+ private static final String HEADER_KEY = "transactionStatus";
+ private static final byte[] ABORTED_VALUE = "aborted".getBytes();
+ private static final byte[] COMMITTED_VALUE = "committed".getBytes();
+
+ @ClusterTest
+ public void testTransactionalProducerSingleBrokerMaxInFlightOne(ClusterInstance clusterInstance) throws InterruptedException {
+ // We want to test with one broker to verify multiple requests queued on a connection
+ assertEquals(1, clusterInstance.brokers().size());
+
+ clusterInstance.createTopic(TOPIC1, 4, (short) 1);
+ clusterInstance.createTopic(TOPIC2, 4, (short) 1);
+
+ try (Producer<byte[], byte[]> producer = clusterInstance.producer(Map.of(
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional-producer",
+ ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1
+ ))
+ ) {
+ producer.initTransactions();
+
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(TOPIC2, null, "2".getBytes(), "2".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, ABORTED_VALUE))));
+ producer.send(new ProducerRecord<>(TOPIC1, null, "4".getBytes(), "4".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, ABORTED_VALUE))));
+ producer.flush();
+ producer.abortTransaction();
+
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(TOPIC1, null, "1".getBytes(), "1".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, COMMITTED_VALUE))));
+ producer.send(new ProducerRecord<>(TOPIC2, null, "3".getBytes(), "3".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, COMMITTED_VALUE))));
+ producer.commitTransaction();
+
+ for (GroupProtocol groupProtocol : clusterInstance.supportedGroupProtocols()) {
+ ArrayList<ConsumerRecord<byte[], byte[]>> consumerRecords = new ArrayList<>();
+ try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(Map.of(
+ ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name(),
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false",
+ ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"
+ )
+ )) {
+ consumer.subscribe(List.of(TOPIC1, TOPIC2));
+ TestUtils.waitForCondition(() -> {
+ ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(100));
+ records.forEach(consumerRecords::add);
+ return consumerRecords.size() == 2;
+ }, 15_000, () -> "Consumer with protocol " + groupProtocol.name + " should consume 2 records, but get " + consumerRecords.size());
Review Comment:
This won’t compile because `groupProtocol.name` is not a field; it should be `groupProtocol.name()`. Also, this test passes `groupProtocol.name()` to `GROUP_PROTOCOL_CONFIG` while other tests in this PR use `name().toLowerCase(Locale.ROOT)`; if the config parser is case-sensitive, this will fail at runtime.
##########
.github/workflows/deflake.yml:
##########
@@ -60,26 +65,52 @@ jobs:
TEST_PATTERN: ${{ inputs.test-pattern }}
run: |
set +e
- ./gradlew --info --build-cache --scan --continue \
+ ./.github/scripts/thread-dump.sh &
+ timeout 60m ./gradlew --info --build-cache --scan --continue \
-PtestLoggingEvents=started,passed,skipped,failed \
- -PignoreFailures=true -PmaxParallelForks=2 \
+ -PmaxParallelForks=2 \
-Pkafka.cluster.test.repeat=$TEST_REPEAT \
-PmaxTestRetries=$TEST_REPEAT -PmaxTestRetryFailures=0 \
- -PmaxQuarantineTestRetries=$TEST_REPEAT -PmaxQuarantineTestRetryFailures=0 \
- ${TEST_MODULE}:test ${TEST_MODULE}:quarantinedTest --tests $TEST_PATTERN
+ ${TEST_MODULE}:test --tests $TEST_PATTERN
exitcode="$?"
echo "exitcode=$exitcode" >> $GITHUB_OUTPUT
- - name: Archive JUnit reports
+
+ - name: Archive JUnit HTML reports
+ uses: actions/upload-artifact@v4
+ id: archive-junit-html
+ with:
+ name: junit-html-reports
+ path: |
+ **/build/reports/tests/*
+ compression-level: 9
+ if-no-files-found: ignore
+
+ - name: Archive JUnit XML
+ uses: actions/upload-artifact@v4
+ with:
+ name: junit-xml
+ path: |
+ build/junit-xml/**/*.xml
+ compression-level: 9
+ if-no-files-found: ignore
+
+ - name: Archive Thread Dumps
+ id: archive-thread-dump
+ if: steps.junit-test.outputs.gradle-exitcode == '124'
Review Comment:
The `junit-test` step writes `exitcode` to `$GITHUB_OUTPUT`, so the output key is `exitcode`, not `gradle-exitcode`. As written, the timeout detection and JUnit parsing will treat the exit code as missing and may fail the workflow incorrectly. Update these references to `steps.junit-test.outputs.exitcode` (or rename the output variable to match).
##########
.github/workflows/generate-reports.yml:
##########
@@ -33,11 +32,12 @@ jobs:
env:
GITHUB_CONTEXT: ${{ toJson(github) }}
- name: Checkout code
- uses: actions/checkout@v4
+ uses: actions/checkout@v5
- name: Setup Python
uses: ./.github/actions/setup-python
- name: Run Report
+ if : ${{ ! github.event.repository.fork }}
Review Comment:
This is invalid GitHub Actions YAML (`if :` with a space before `:`). It should be `if:`. As-is, the workflow will fail to load/execute.
```suggestion
if: ${{ ! github.event.repository.fork }}
```
##########
.github/dependabot.yml:
##########
@@ -0,0 +1,26 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+version: 2
+updates:
+ - package-ecosystem: "github-actions"
+ directory: "/"
+ schedule:
+ # 'daily' only runs on weekdays
+ interval: "cron"
+ cronjob: "00 20 * * *"
+ open-pull-requests-limit: 10
+ cooldown:
+ default: 4
Review Comment:
This does not match Dependabot’s supported configuration schema. `schedule.interval` does not support `cron`, and keys like `cronjob` and `cooldown` are not valid in Dependabot config. This will likely cause Dependabot parsing errors and prevent updates from running; please switch to supported fields (e.g., `interval: daily/weekly/monthly`, plus `time`/`timezone` if needed).
```suggestion
interval: "daily"
open-pull-requests-limit: 10
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/RackAwareAutoTopicCreationTest.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.server.config.ReplicationConfigs;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.server.config.ServerLogConfigs;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@ClusterTestDefaults(types = {Type.KRAFT},
+ brokers = 4,
+ serverProperties = {
+ @ClusterConfigProperty(key = ServerLogConfigs.NUM_PARTITIONS_CONFIG, value = "8"),
+ @ClusterConfigProperty(key = ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, value = "2"),
+ @ClusterConfigProperty(id = 0, key = ServerConfigs.BROKER_RACK_CONFIG, value = "0"),
+ @ClusterConfigProperty(id = 1, key = ServerConfigs.BROKER_RACK_CONFIG, value = "0"),
+ @ClusterConfigProperty(id = 2, key = ServerConfigs.BROKER_RACK_CONFIG, value = "1"),
+ @ClusterConfigProperty(id = 3, key = ServerConfigs.BROKER_RACK_CONFIG, value = "1"),
+ })
+public class RackAwareAutoTopicCreationTest {
+
+ private static final String TOPIC = "topic";
+
+ @ClusterTest
+ public void testAutoCreateTopic(ClusterInstance cluster) throws Exception {
+
+ try (Admin admin = cluster.admin();
+ Producer<byte[], byte[]> producer = cluster.producer()) {
+
+ // send a record to trigger auto create topic
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(TOPIC, null, "key".getBytes(), "value".getBytes());
+ assertEquals(0L, producer.send(record).get().offset(), "Should have offset 0");
+
+ // check broker rack content
+ Map<Integer, String> expectedBrokerToRackMap = Map.of(
+ 0, "0",
+ 1, "0",
+ 2, "1",
+ 3, "1"
+ );
+ Map<Integer, String> actualBrokerToRackMap = getBrokerToRackMap(cluster);
+ assertEquals(expectedBrokerToRackMap, actualBrokerToRackMap);
+
+ // get topic assignments and check it's content
+ Map<Integer, List<Integer>> assignments = getTopicAssignment(admin);
+ for (List<Integer> brokerList : assignments.values()) {
+ assertEquals(new HashSet<>(brokerList).size(), brokerList.size(),
+ "More than one replica is assigned to same broker for the same partition");
+ }
+
+ // check rack count for each partition
+ ReplicaDistributions distribution = getReplicaDistribution(assignments, expectedBrokerToRackMap);
+ Map<String, String> serverProperties = cluster.config().serverProperties();
+ int numPartition = Integer.parseInt(serverProperties.get(ServerLogConfigs.NUM_PARTITIONS_CONFIG));
+ int replicationFactor = Integer.parseInt(serverProperties.get(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG));
+ List<Integer> expectedRackCounts = Collections.nCopies(numPartition, replicationFactor);
+ List<Integer> actualRackCounts = distribution.partitionToRackMap.values().stream()
+ .map(racks -> (int) racks.stream().distinct().count())
+ .collect(Collectors.toList());
+ assertEquals(expectedRackCounts, actualRackCounts,
+ "More than one replica of the same partition is assigned to the same rack");
+
+ // check replica count for each partition
+ int numBrokers = cluster.brokers().size();
+ int numReplicasPerBroker = numPartition * replicationFactor / numBrokers;
+ List<Integer> expectedReplicasCounts = Collections.nCopies(numBrokers, numReplicasPerBroker);
+ List<Integer> actualReplicasCounts = new ArrayList<>(distribution.partitionToCountMap.values());
+ assertEquals(expectedReplicasCounts, actualReplicasCounts, "Replica count is not even for broker");
+ }
+ }
+
+ private static Map<Integer, List<Integer>> getTopicAssignment(Admin admin) throws Exception {
+ TopicDescription topicDescription = admin.describeTopics(List.of(TOPIC)).allTopicNames().get().get(TOPIC);
+ return topicDescription.partitions().stream()
+ .collect(Collectors.toMap(
+ TopicPartitionInfo::partition,
+ p -> p.replicas().stream().map(Node::id).collect(Collectors.toList())));
+ }
+
+ private static Map<Integer, String> getBrokerToRackMap(ClusterInstance cluster) throws Exception {
+ Map<Integer, String> actualBrokerToRackMap = new HashMap<>();
+ try (Admin admin = cluster.admin()) {
+ admin.describeCluster().nodes().get().forEach(node -> actualBrokerToRackMap.put(node.id(), node.rack()));
+ }
+ return actualBrokerToRackMap;
+ }
+
+
+ private static ReplicaDistributions getReplicaDistribution(Map<Integer, List<Integer>> assignment,
+ Map<Integer, String> brokerRackMapping) {
+ Map<Integer, List<String>> partitionToRackMap = new HashMap<>();
+ Map<Integer, Integer> partitionToCountMap = new HashMap<>();
+
+ for (Map.Entry<Integer, List<Integer>> entry : assignment.entrySet()) {
+ int partitionId = entry.getKey();
+ List<Integer> replicaList = entry.getValue();
+
+ for (int brokerId : replicaList) {
+ partitionToCountMap.put(brokerId, partitionToCountMap.getOrDefault(brokerId, 0) + 1);
Review Comment:
`partitionToCountMap` is keyed by `brokerId`, not `partitionId`, so the name is misleading. Rename to something like `brokerToReplicaCount` to reflect its contents and avoid confusion.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCloseTest.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+
+import java.time.Duration;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.kafka.clients.ClientsTestUtils.BaseConsumerTestcase.BROKER_COUNT;
+import static org.apache.kafka.clients.ClientsTestUtils.consumeRecords;
+import static org.apache.kafka.clients.ClientsTestUtils.sendRecords;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.DEFAULT_FETCH_MAX_WAIT_MS;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_PROTOCOL_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = BROKER_COUNT,
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, value = "60000"),
+ @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "10"),
+ }
+)
+public class PlaintextConsumerCloseTest {
+
+ private final ClusterInstance cluster;
+
+ public PlaintextConsumerCloseTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ @ClusterTest
+ public void testClassicConsumerCloseWithDefaultTakesAtLeastFetchMaxWaitMs() throws Exception {
+ testCloseWithDefaultTakesAtLeastFetchMaxWaitMs(GroupProtocol.CLASSIC);
+ }
+
+ @ClusterTest
+ public void testAsyncConsumerCloseWithDefaultTakesAtLeastFetchMaxWaitMs() throws Exception {
+ testCloseWithDefaultTakesAtLeastFetchMaxWaitMs(GroupProtocol.CONSUMER);
+ }
+
+ private void testCloseWithDefaultTakesAtLeastFetchMaxWaitMs(GroupProtocol groupProtocol) throws Exception {
+ long closeMs = calculateConsumerCloseDelay(groupProtocol, Consumer::close);
+ assertTrue(
+ closeMs >= DEFAULT_FETCH_MAX_WAIT_MS,
+ "Closing a consumer with the default close() should take longer than " + DEFAULT_FETCH_MAX_WAIT_MS + " ms, but actually took " + closeMs + " ms"
+ );
+ }
+
+ @ClusterTest
+ public void testClassicConsumerCloseWithTimeoutIgnoresFetchMaxWaitMs() throws Exception {
+ testCloseWithTimeoutIgnoresFetchMaxWaitMs(GroupProtocol.CLASSIC);
+ }
+
+ @ClusterTest
+ public void testAsyncConsumerCloseWithTimeoutIgnoresFetchMaxWaitMs() throws Exception {
+ testCloseWithTimeoutIgnoresFetchMaxWaitMs(GroupProtocol.CONSUMER);
+ }
+
+ private void testCloseWithTimeoutIgnoresFetchMaxWaitMs(GroupProtocol groupProtocol) throws Exception {
+ long timeoutMs = 100;
+
+ // Close the Consumer with a specified timeout that's much shorter than the default fetch timeout
+ // to ensure the fetch.max.wait.ms is effectively ignored.
+ long closeMs = calculateConsumerCloseDelay(
+ groupProtocol,
+ c -> c.close(CloseOptions.timeout(Duration.ofMillis(timeoutMs)))
+ );
+ assertTrue(
+ closeMs <= DEFAULT_FETCH_MAX_WAIT_MS,
+ "Closing a consumer with a timeout of " + timeoutMs + " ms should take less than " + DEFAULT_FETCH_MAX_WAIT_MS + " ms, but actually took " + closeMs + " ms"
+ );
+ }
+
+ private long calculateConsumerCloseDelay(GroupProtocol groupProtocol,
+ java.util.function.Consumer<Consumer<byte[], byte[]>> closeOperation) throws Exception {
+ Map<String, Object> consumerConfig = Map.of(
+ GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT),
+ KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName(),
+ VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName(),
+ AUTO_OFFSET_RESET_CONFIG, "earliest",
+ GROUP_ID_CONFIG, "group_test",
+ BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()
+ );
+ var topicName = "calculate-consumer-close-delay";
+ var numRecords = 100;
+ cluster.createTopic(topicName, 2, (short) 2, Map.of());
+
+ sendRecords(cluster, new TopicPartition(topicName, 0), numRecords);
+
+ try (Consumer<byte[], byte[]> consumer = cluster.consumer(consumerConfig)) {
+ consumer.subscribe(Set.of(topicName));
+ consumeRecords(consumer, numRecords);
+
+ long start = System.currentTimeMillis();
+ closeOperation.accept(consumer);
+ long end = System.currentTimeMillis();
+ return end - start;
Review Comment:
This measures the time for an explicit `closeOperation.accept(consumer)` while also relying on try-with-resources to close again. The implicit close runs after computing the return value but before returning, which can make the test assertions misleading and can double-close the consumer. To make the timing meaningful and avoid double-close side-effects, don’t use try-with-resources when the close is part of the test (or move the timing measurement outside and ensure the consumer is only closed once).
```suggestion
Consumer<byte[], byte[]> consumer = cluster.consumer(consumerConfig);
boolean closed = false;
try {
consumer.subscribe(Set.of(topicName));
consumeRecords(consumer, numRecords);
long start = System.currentTimeMillis();
closeOperation.accept(consumer);
long end = System.currentTimeMillis();
closed = true;
return end - start;
} finally {
if (!closed) {
consumer.close();
}
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerWithLegacyMessageFormatIntegrationTest.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.clients.ClientsTestUtils;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.record.internal.AbstractRecords;
+import org.apache.kafka.common.record.internal.CompressionType;
+import org.apache.kafka.common.record.internal.MemoryRecords;
+import org.apache.kafka.common.record.internal.MemoryRecordsBuilder;
+import org.apache.kafka.common.record.internal.RecordBatch;
+import org.apache.kafka.common.record.internal.RecordVersion;
+import org.apache.kafka.common.record.internal.SimpleRecord;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.storage.internals.log.UnifiedLog;
+
+import org.junit.jupiter.api.BeforeEach;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_PROTOCOL_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+
+@ClusterTestDefaults(
+ brokers = 3
+)
+public class ConsumerWithLegacyMessageFormatIntegrationTest {
+
+ private final ClusterInstance cluster;
+
+ private final String topic1 = "part-test-topic-1";
+ private final String topic2 = "part-test-topic-2";
+ private final String topic3 = "part-test-topic-3";
+
+ private final TopicPartition t1p0 = new TopicPartition(topic1, 0);
+ private final TopicPartition t1p1 = new TopicPartition(topic1, 1);
+ private final TopicPartition t2p0 = new TopicPartition(topic2, 0);
+ private final TopicPartition t2p1 = new TopicPartition(topic2, 1);
+ private final TopicPartition t3p0 = new TopicPartition(topic3, 0);
+ private final TopicPartition t3p1 = new TopicPartition(topic3, 1);
+
+ public ConsumerWithLegacyMessageFormatIntegrationTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ private void appendLegacyRecords(int numRecords, TopicPartition tp, int brokerId, byte magicValue) {
+ List<SimpleRecord> records = new ArrayList<>();
+ for (int i = 0; i < numRecords; i++) {
+ records.add(new SimpleRecord(i, ("key " + i).getBytes(), ("value " + i).getBytes()));
+ }
+
+ ByteBuffer buffer = ByteBuffer.allocate(AbstractRecords.estimateSizeInBytes(magicValue,
+ CompressionType.NONE, records));
+ MemoryRecordsBuilder builder = MemoryRecords.builder(
+ buffer,
+ magicValue,
+ Compression.of(CompressionType.NONE).build(),
+ TimestampType.CREATE_TIME,
+ 0L,
+ RecordBatch.NO_TIMESTAMP,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ 0,
+ false,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH
+ );
+
+ records.forEach(builder::append);
+
+ cluster.brokers().values().stream()
+ .filter(b -> b.config().brokerId() == brokerId)
+ .forEach(b -> {
+ UnifiedLog unifiedLog = b.replicaManager().logManager().getLog(tp, false).get();
+ unifiedLog.appendAsLeaderWithRecordVersion(builder.build(), 0, RecordVersion.lookup(magicValue));
+ // Default isolation.level is read_uncommitted. It makes Partition#fetchOffsetForTimestamp to return UnifiedLog#highWatermark,
+ // so increasing high watermark to make it return the correct offset.
+ assertDoesNotThrow(() -> unifiedLog.maybeIncrementHighWatermark(unifiedLog.logEndOffsetMetadata()));
+ });
+ }
+
+ private void createTopicWithAssignment(String topic, Map<Integer, List<Integer>> assignment) throws InterruptedException {
+ try (Admin admin = cluster.admin()) {
+ NewTopic newTopic = new NewTopic(topic, assignment);
+ admin.createTopics(List.of(newTopic));
+ cluster.waitTopicCreation(topic, assignment.size());
+ }
+ }
+
+ @BeforeEach
+ public void setupTopics() throws InterruptedException {
+ cluster.createTopic(topic1, 2, (short) 1);
+ createTopicWithAssignment(topic2, Map.of(0, List.of(0), 1, List.of(1)));
+ createTopicWithAssignment(topic3, Map.of(0, List.of(0), 1, List.of(1)));
+
+ // v2 message format for topic1
+ ClientsTestUtils.sendRecords(cluster, t1p0, 100, 0);
+ ClientsTestUtils.sendRecords(cluster, t1p1, 100, 0);
+ // v0 message format for topic2
+ appendLegacyRecords(100, t2p0, 0, RecordBatch.MAGIC_VALUE_V0);
+ appendLegacyRecords(100, t2p1, 1, RecordBatch.MAGIC_VALUE_V0);
+ // v1 message format for topic3
+ appendLegacyRecords(100, t3p0, 0, RecordBatch.MAGIC_VALUE_V1);
+ appendLegacyRecords(100, t3p1, 1, RecordBatch.MAGIC_VALUE_V1);
+ }
+
+ @ClusterTest
+ public void testOffsetsForTimesWithClassicConsumer() {
+ testOffsetsForTimes(GroupProtocol.CLASSIC);
+ }
+
+ @ClusterTest
+ public void testOffsetsForTimesWithAsyncConsumer() {
+ testOffsetsForTimes(GroupProtocol.CONSUMER);
+ }
+
+ public void testOffsetsForTimes(GroupProtocol groupProtocol) {
+ try (Consumer<Object, Object> consumer = cluster.consumer(Map.of(
+ GROUP_PROTOCOL_CONFIG, groupProtocol.name.toLowerCase(Locale.ROOT)))
Review Comment:
`GroupProtocol` is an enum; `name` is not a field. This should call `groupProtocol.name()` (and typically lower-case it) to compile. Please update this to use `groupProtocol.name().toLowerCase(Locale.ROOT)`.
```suggestion
GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT)))
```
##########
.github/workflows/ci.yml:
##########
@@ -19,6 +19,7 @@ on:
push:
branches:
- 'trunk'
+ - '4.3'
Review Comment:
The PR description says this change is about testing Gradle report upload when JDK versions differ between trunk and 4.0, but the diff includes broad CI/workflow changes, many new integration tests, Checkstyle refactors, and NOTICE/LICENSE edits. Please update the PR description/title to reflect the actual scope, or split this into smaller PRs aligned with a single purpose.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/TransactionsWithMaxInFlightOneTest.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
+import org.apache.kafka.coordinator.transaction.TransactionStateManagerConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.test.TestUtils;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ClusterTestDefaults(
+ types = {Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "false"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = "log.unclean.leader.election.enable", value = "false"),
+ @ClusterConfigProperty(key = ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG, value = "false"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "0"),
+ @ClusterConfigProperty(key = TransactionStateManagerConfig.TRANSACTIONS_ABORT_TIMED_OUT_TRANSACTION_CLEANUP_INTERVAL_MS_CONFIG, value = "200")
+ }
+)
+public class TransactionsWithMaxInFlightOneTest {
+ private static final String TOPIC1 = "topic1";
+ private static final String TOPIC2 = "topic2";
+ private static final String HEADER_KEY = "transactionStatus";
+ private static final byte[] ABORTED_VALUE = "aborted".getBytes();
+ private static final byte[] COMMITTED_VALUE = "committed".getBytes();
+
+ @ClusterTest
+ public void testTransactionalProducerSingleBrokerMaxInFlightOne(ClusterInstance clusterInstance) throws InterruptedException {
+ // We want to test with one broker to verify multiple requests queued on a connection
+ assertEquals(1, clusterInstance.brokers().size());
+
+ clusterInstance.createTopic(TOPIC1, 4, (short) 1);
+ clusterInstance.createTopic(TOPIC2, 4, (short) 1);
+
+ try (Producer<byte[], byte[]> producer = clusterInstance.producer(Map.of(
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactional-producer",
+ ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1
+ ))
+ ) {
+ producer.initTransactions();
+
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(TOPIC2, null, "2".getBytes(), "2".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, ABORTED_VALUE))));
+ producer.send(new ProducerRecord<>(TOPIC1, null, "4".getBytes(), "4".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, ABORTED_VALUE))));
+ producer.flush();
+ producer.abortTransaction();
+
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(TOPIC1, null, "1".getBytes(), "1".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, COMMITTED_VALUE))));
+ producer.send(new ProducerRecord<>(TOPIC2, null, "3".getBytes(), "3".getBytes(), Collections.singleton(new RecordHeader(HEADER_KEY, COMMITTED_VALUE))));
+ producer.commitTransaction();
+
+ for (GroupProtocol groupProtocol : clusterInstance.supportedGroupProtocols()) {
+ ArrayList<ConsumerRecord<byte[], byte[]>> consumerRecords = new ArrayList<>();
+ try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(Map.of(
+ ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name(),
Review Comment:
This won’t compile because `groupProtocol.name` is not a field; it should be `groupProtocol.name()`. Also, this test passes `groupProtocol.name()` to `GROUP_PROTOCOL_CONFIG` while other tests in this PR use `name().toLowerCase(Locale.ROOT)`; if the config parser is case-sensitive, this will fail at runtime.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCloseTest.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+
+import java.time.Duration;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.kafka.clients.ClientsTestUtils.BaseConsumerTestcase.BROKER_COUNT;
+import static org.apache.kafka.clients.ClientsTestUtils.consumeRecords;
+import static org.apache.kafka.clients.ClientsTestUtils.sendRecords;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.DEFAULT_FETCH_MAX_WAIT_MS;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_PROTOCOL_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = BROKER_COUNT,
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, value = "60000"),
+ @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "10"),
+ }
+)
+public class PlaintextConsumerCloseTest {
+
+ private final ClusterInstance cluster;
+
+ public PlaintextConsumerCloseTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ @ClusterTest
+ public void testClassicConsumerCloseWithDefaultTakesAtLeastFetchMaxWaitMs() throws Exception {
+ testCloseWithDefaultTakesAtLeastFetchMaxWaitMs(GroupProtocol.CLASSIC);
+ }
+
+ @ClusterTest
+ public void testAsyncConsumerCloseWithDefaultTakesAtLeastFetchMaxWaitMs() throws Exception {
+ testCloseWithDefaultTakesAtLeastFetchMaxWaitMs(GroupProtocol.CONSUMER);
+ }
+
+ private void testCloseWithDefaultTakesAtLeastFetchMaxWaitMs(GroupProtocol groupProtocol) throws Exception {
+ long closeMs = calculateConsumerCloseDelay(groupProtocol, Consumer::close);
+ assertTrue(
+ closeMs >= DEFAULT_FETCH_MAX_WAIT_MS,
+ "Closing a consumer with the default close() should take longer than " + DEFAULT_FETCH_MAX_WAIT_MS + " ms, but actually took " + closeMs + " ms"
+ );
+ }
+
+ @ClusterTest
+ public void testClassicConsumerCloseWithTimeoutIgnoresFetchMaxWaitMs() throws Exception {
+ testCloseWithTimeoutIgnoresFetchMaxWaitMs(GroupProtocol.CLASSIC);
+ }
+
+ @ClusterTest
+ public void testAsyncConsumerCloseWithTimeoutIgnoresFetchMaxWaitMs() throws Exception {
+ testCloseWithTimeoutIgnoresFetchMaxWaitMs(GroupProtocol.CONSUMER);
+ }
+
+ private void testCloseWithTimeoutIgnoresFetchMaxWaitMs(GroupProtocol groupProtocol) throws Exception {
+ long timeoutMs = 100;
+
+ // Close the Consumer with a specified timeout that's much shorter than the default fetch timeout
+ // to ensure the fetch.max.wait.ms is effectively ignored.
+ long closeMs = calculateConsumerCloseDelay(
+ groupProtocol,
+ c -> c.close(CloseOptions.timeout(Duration.ofMillis(timeoutMs)))
+ );
+ assertTrue(
+ closeMs <= DEFAULT_FETCH_MAX_WAIT_MS,
+ "Closing a consumer with a timeout of " + timeoutMs + " ms should take less than " + DEFAULT_FETCH_MAX_WAIT_MS + " ms, but actually took " + closeMs + " ms"
+ );
+ }
+
+ private long calculateConsumerCloseDelay(GroupProtocol groupProtocol,
+ java.util.function.Consumer<Consumer<byte[], byte[]>> closeOperation) throws Exception {
+ Map<String, Object> consumerConfig = Map.of(
+ GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT),
+ KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName(),
+ VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName(),
+ AUTO_OFFSET_RESET_CONFIG, "earliest",
+ GROUP_ID_CONFIG, "group_test",
+ BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()
+ );
+ var topicName = "calculate-consumer-close-delay";
+ var numRecords = 100;
+ cluster.createTopic(topicName, 2, (short) 2, Map.of());
+
+ sendRecords(cluster, new TopicPartition(topicName, 0), numRecords);
+
+ try (Consumer<byte[], byte[]> consumer = cluster.consumer(consumerConfig)) {
+ consumer.subscribe(Set.of(topicName));
+ consumeRecords(consumer, numRecords);
+
+ long start = System.currentTimeMillis();
+ closeOperation.accept(consumer);
+ long end = System.currentTimeMillis();
+ return end - start;
Review Comment:
This measures the time for an explicit `closeOperation.accept(consumer)` while also relying on try-with-resources to close again. The implicit close runs after computing the return value but before returning, which can make the test assertions misleading and can double-close the consumer. To make the timing meaningful and avoid double-close side-effects, don’t use try-with-resources when the close is part of the test (or move the timing measurement outside and ensure the consumer is only closed once).
```suggestion
Consumer<byte[], byte[]> consumer = cluster.consumer(consumerConfig);
boolean closed = false;
try {
consumer.subscribe(Set.of(topicName));
consumeRecords(consumer, numRecords);
long start = System.currentTimeMillis();
closeOperation.accept(consumer);
closed = true;
long end = System.currentTimeMillis();
return end - start;
} finally {
if (!closed) {
consumer.close();
}
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/RackAwareAutoTopicCreationTest.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.server.config.ReplicationConfigs;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.server.config.ServerLogConfigs;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@ClusterTestDefaults(types = {Type.KRAFT},
+ brokers = 4,
+ serverProperties = {
+ @ClusterConfigProperty(key = ServerLogConfigs.NUM_PARTITIONS_CONFIG, value = "8"),
+ @ClusterConfigProperty(key = ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG, value = "2"),
+ @ClusterConfigProperty(id = 0, key = ServerConfigs.BROKER_RACK_CONFIG, value = "0"),
+ @ClusterConfigProperty(id = 1, key = ServerConfigs.BROKER_RACK_CONFIG, value = "0"),
+ @ClusterConfigProperty(id = 2, key = ServerConfigs.BROKER_RACK_CONFIG, value = "1"),
+ @ClusterConfigProperty(id = 3, key = ServerConfigs.BROKER_RACK_CONFIG, value = "1"),
+ })
+public class RackAwareAutoTopicCreationTest {
+
+ private static final String TOPIC = "topic";
+
+ @ClusterTest
+ public void testAutoCreateTopic(ClusterInstance cluster) throws Exception {
+
+ try (Admin admin = cluster.admin();
+ Producer<byte[], byte[]> producer = cluster.producer()) {
+
+ // send a record to trigger auto create topic
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(TOPIC, null, "key".getBytes(), "value".getBytes());
+ assertEquals(0L, producer.send(record).get().offset(), "Should have offset 0");
+
+ // check broker rack content
+ Map<Integer, String> expectedBrokerToRackMap = Map.of(
+ 0, "0",
+ 1, "0",
+ 2, "1",
+ 3, "1"
+ );
+ Map<Integer, String> actualBrokerToRackMap = getBrokerToRackMap(cluster);
+ assertEquals(expectedBrokerToRackMap, actualBrokerToRackMap);
+
+ // get topic assignments and check it's content
+ Map<Integer, List<Integer>> assignments = getTopicAssignment(admin);
+ for (List<Integer> brokerList : assignments.values()) {
+ assertEquals(new HashSet<>(brokerList).size(), brokerList.size(),
+ "More than one replica is assigned to same broker for the same partition");
+ }
+
+ // check rack count for each partition
+ ReplicaDistributions distribution = getReplicaDistribution(assignments, expectedBrokerToRackMap);
+ Map<String, String> serverProperties = cluster.config().serverProperties();
+ int numPartition = Integer.parseInt(serverProperties.get(ServerLogConfigs.NUM_PARTITIONS_CONFIG));
+ int replicationFactor = Integer.parseInt(serverProperties.get(ReplicationConfigs.DEFAULT_REPLICATION_FACTOR_CONFIG));
+ List<Integer> expectedRackCounts = Collections.nCopies(numPartition, replicationFactor);
+ List<Integer> actualRackCounts = distribution.partitionToRackMap.values().stream()
+ .map(racks -> (int) racks.stream().distinct().count())
+ .collect(Collectors.toList());
+ assertEquals(expectedRackCounts, actualRackCounts,
+ "More than one replica of the same partition is assigned to the same rack");
+
+ // check replica count for each partition
+ int numBrokers = cluster.brokers().size();
+ int numReplicasPerBroker = numPartition * replicationFactor / numBrokers;
+ List<Integer> expectedReplicasCounts = Collections.nCopies(numBrokers, numReplicasPerBroker);
+ List<Integer> actualReplicasCounts = new ArrayList<>(distribution.partitionToCountMap.values());
+ assertEquals(expectedReplicasCounts, actualReplicasCounts, "Replica count is not even for broker");
+ }
+ }
+
+ private static Map<Integer, List<Integer>> getTopicAssignment(Admin admin) throws Exception {
+ TopicDescription topicDescription = admin.describeTopics(List.of(TOPIC)).allTopicNames().get().get(TOPIC);
+ return topicDescription.partitions().stream()
+ .collect(Collectors.toMap(
+ TopicPartitionInfo::partition,
+ p -> p.replicas().stream().map(Node::id).collect(Collectors.toList())));
+ }
+
+ private static Map<Integer, String> getBrokerToRackMap(ClusterInstance cluster) throws Exception {
+ Map<Integer, String> actualBrokerToRackMap = new HashMap<>();
+ try (Admin admin = cluster.admin()) {
+ admin.describeCluster().nodes().get().forEach(node -> actualBrokerToRackMap.put(node.id(), node.rack()));
+ }
+ return actualBrokerToRackMap;
+ }
+
+
+ private static ReplicaDistributions getReplicaDistribution(Map<Integer, List<Integer>> assignment,
+ Map<Integer, String> brokerRackMapping) {
+ Map<Integer, List<String>> partitionToRackMap = new HashMap<>();
+ Map<Integer, Integer> partitionToCountMap = new HashMap<>();
Review Comment:
`partitionToCountMap` is keyed by `brokerId`, not `partitionId`, so the name is misleading. Rename to something like `brokerToReplicaCount` to reflect its contents and avoid confusion.
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ConsumerAssignmentPoller.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.server.util.ShutdownableThread;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+public class ConsumerAssignmentPoller extends ShutdownableThread {
+ private final Consumer<byte[], byte[]> consumer;
+ private final Set<TopicPartition> partitionsToAssign;
+ private volatile Optional<Throwable> thrownException = Optional.empty();
+ private volatile int receivedMessages = 0;
+ private final Set<TopicPartition> partitionAssignment = Collections.synchronizedSet(new HashSet<>());
+ private volatile boolean subscriptionChanged = false;
+ private List<String> topicsSubscription;
+ private final ConsumerRebalanceListener rebalanceListener;
+
+ public ConsumerAssignmentPoller(Consumer<byte[], byte[]> consumer, List<String> topicsToSubscribe) {
+ this(consumer, topicsToSubscribe, Set.of(), null);
+ }
+
+ public ConsumerAssignmentPoller(Consumer<byte[], byte[]> consumer, Set<TopicPartition> partitionsToAssign) {
+ this(consumer, List.of(), partitionsToAssign, null);
+ }
+
+ public ConsumerAssignmentPoller(Consumer<byte[], byte[]> consumer,
+ List<String> topicsToSubscribe,
+ Set<TopicPartition> partitionsToAssign,
+ ConsumerRebalanceListener userRebalanceListener) {
+ super("daemon-consumer-assignment", false);
+ this.consumer = consumer;
+ this.partitionsToAssign = partitionsToAssign;
+ this.topicsSubscription = topicsToSubscribe;
+
+ this.rebalanceListener = new ConsumerRebalanceListener() {
+ @Override
+ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+ partitionAssignment.addAll(partitions);
+ if (userRebalanceListener != null)
+ userRebalanceListener.onPartitionsAssigned(partitions);
+ }
+
+ @Override
+ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+ partitionAssignment.removeAll(partitions);
+ if (userRebalanceListener != null)
+ userRebalanceListener.onPartitionsRevoked(partitions);
+ }
+ };
+
+ if (partitionsToAssign.isEmpty()) {
+ consumer.subscribe(topicsToSubscribe, rebalanceListener);
+ } else {
+ consumer.assign(List.copyOf(partitionsToAssign));
+ }
+ }
+
+ public Set<TopicPartition> consumerAssignment() {
+ return Set.copyOf(partitionAssignment);
Review Comment:
`partitionAssignment` is a `Collections.synchronizedSet(...)`, but iteration/copying still requires external synchronization. `Set.copyOf(partitionAssignment)` can throw `ConcurrentModificationException` while rebalance callbacks mutate it. Fix by synchronizing on `partitionAssignment` while copying (e.g., `synchronized (partitionAssignment) { return Set.copyOf(partitionAssignment); }`) or switch to a concurrent set implementation.
```suggestion
synchronized (partitionAssignment) {
return Set.copyOf(partitionAssignment);
}
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/ConcurrentListOffsetsRequestTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.DefaultHostResolver;
+import org.apache.kafka.clients.NetworkClient;
+import org.apache.kafka.clients.admin.internals.PartitionLeaderCache;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = 3
+)
+public class ConcurrentListOffsetsRequestTest {
+ private static final String TOPIC = "topic";
+ private static final short REPLICAS = 1;
+ private static final int PARTITION = 2;
+ private static final int TIMEOUT = 1000;
+ private final ClusterInstance clusterInstance;
+ private Admin adminClient;
+ private NetworkClient networkClient;
+ private final AtomicBoolean injectHostResolverError = new AtomicBoolean(false);
+
+ ConcurrentListOffsetsRequestTest(ClusterInstance clusterInstance) {
+ this.clusterInstance = clusterInstance;
+ }
+
+ @BeforeEach
+ public void setup() throws Exception {
+ clusterInstance.waitForReadyBrokers();
+ clusterInstance.createTopic(TOPIC, PARTITION, REPLICAS);
+ Map<String, Object> props = Map.of(
+ "default.api.timeout.ms", TIMEOUT,
+ "request.timeout.ms", TIMEOUT,
+ CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers());
+ adminClient = KafkaAdminClient.createInternal(new AdminClientConfig(clusterInstance.setClientSaslConfig(props), true),
+ null, new TestHostResolver());
+
+ networkClient = TestUtils.fieldValue(adminClient, KafkaAdminClient.class, "client");
+ }
+
+ @AfterEach
+ public void teardown() {
+ Utils.closeQuietly(adminClient, "ListOffsetsAdminClient");
+ }
+
+ @ClusterTest
+ public void correctlyHandleConcurrentModificationOfPartitionLeaderCache() throws Exception {
+ // making one request to prepopulate the partition leader cache so we have something to delete later
+ listAllOffsets().all().get(TIMEOUT * 2, TimeUnit.SECONDS);
Review Comment:
`TIMEOUT` is defined in milliseconds (`1000`), but this uses `TimeUnit.SECONDS`, turning a 2s wait into ~2000s. This can massively slow/hang CI. Use `TimeUnit.MILLISECONDS` here (or rename the constant to reflect seconds if that’s intended).
```suggestion
listAllOffsets().all().get(TIMEOUT * 2, TimeUnit.MILLISECONDS);
```
--
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