Skip to content

MINOR: add test for kafka-consumer-groups.sh should not fail when partition offline #20235

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AndrewJSchofield merged 19 commits into from
Jul 31, 2025
Merged

MINOR: add test for kafka-consumer-groups.sh should not fail when partition offline #20235

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
13c821c
init integrationTest
TaiJuWu Jul 24, 2025
a0fdb86
use delete dir to create offline partition
TaiJuWu Jul 24, 2025
f7a4b03
Revert "use delete dir to create offline partition"
TaiJuWu Jul 24, 2025
a18d896
add integration test
TaiJuWu Jul 24, 2025
c2a24c9
fix
TaiJuWu Jul 24, 2025
b0e292f
address indentation comment
TaiJuWu Jul 24, 2025
391326a
fix flaky
TaiJuWu Jul 24, 2025
e1cfa54
address Yunyung comment
TaiJuWu Jul 24, 2025
e13bc80
improve performance
TaiJuWu Jul 24, 2025
05b504b
wait shutdown for testResetOffsetsWithPartitionNoneLeader
TaiJuWu Jul 24, 2025
ca27cea
Merge branch 'trunk' into fix_leaderNotFind
TaiJuWu Jul 24, 2025
7cd0703
Revert "wait shutdown for testResetOffsetsWithPartitionNoneLeader"
TaiJuWu Jul 24, 2025
cbe8138
remove wait
TaiJuWu Jul 24, 2025
3406b22
address chia comments
TaiJuWu Jul 24, 2025
b843e4e
set min_in_sync
TaiJuWu Jul 25, 2025
ae74eb4
Merge branch 'trunk' into fix_leaderNotFind
TaiJuWu Jul 25, 2025
5d1c22d
Merge branch 'trunk' into fix_leaderNotFind
TaiJuWu Jul 30, 2025
1a9e642
address chia comments
TaiJuWu Jul 30, 2025
cda748f
fix build
TaiJuWu Jul 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -653,7 +653,7 @@ private List<TopicPartition> filterNoneLeaderPartitions(Collection<TopicPartitio
return adminClient.describeTopics(topics).allTopicNames().get().entrySet()
.stream()
.flatMap(entry -> entry.getValue().partitions().stream()
.filter(partitionInfo -> partitionInfo.leader() == null)
.filter(partitionInfo -> topicPartitions.contains(new TopicPartition(entry.getKey(), partitionInfo.partition())) && partitionInfo.leader() == null)

@Yunyung Yunyung Jul 24, 2025

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: check partitionInfo.leader() == null before creating the TopicPartition for performance.

@TaiJuWu TaiJuWu Jul 24, 2025

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, thanks.

.map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))

@chia7712 chia7712 Jul 24, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add the filter after map to avoid creating extra TopicPartition object.

                                .filter(partitionInfo -> partitionInfo.leader() == null)
                                .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition()))
                                .filter(topicPartitions::contains))

.toList();
} catch (Exception e) {
Expand Down Expand Up @@ -1054,7 +1054,7 @@ private List<TopicPartition> filterNonExistentPartitions(Collection<TopicPartiti
.map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))
.toList();

return topicPartitions.stream().filter(element -> !existPartitions.contains(element)).toList();
return topicPartitions.stream().filter(tp -> !existPartitions.contains(tp)).toList();
Comment thread
TaiJuWu marked this conversation as resolved.
Outdated
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException(e);
}
Expand Down
32 changes: 32 additions & 0 deletions tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.kafka.tools.consumer.group;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.NewPartitionReassignment;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.GroupProtocol;
import org.apache.kafka.clients.consumer.KafkaConsumer;
Expand All @@ -26,7 +27,9 @@
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.errors.LeaderNotAvailableException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.serialization.ByteArraySerializer;
Expand All @@ -51,7 +54,9 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import java.util.function.Supplier;
Expand Down Expand Up @@ -153,6 +158,33 @@ public void testResetOffsetsNotExistingGroup(ClusterInstance cluster) throws Exc
}
}

@ClusterTest(brokers = 5)

@Yunyung Yunyung Jul 24, 2025 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we could just use 2 brokers for this test? 5 is overkill.

@TaiJuWu TaiJuWu Jul 24, 2025 •
edited
Loading

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for review.
I use 3 broker as default since we need to ensure __consumere_offset always have two brokers in any condition and we need to shutdown a broker.

If the replica of __consumer_offset is 1, we can't sure the leader is not shutdown broker.

@chia7712 chia7712 Jul 25, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can two brokers with two replicas fix the issue you mentioned?

@TaiJuWu TaiJuWu Jul 25, 2025 •
edited
Loading

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Change to two brokers and explicitly define min_insyn_replica equals one.

public void testResetOffsetsWithOfflinePartition(ClusterInstance cluster) throws Exception {

@Yunyung Yunyung Jul 24, 2025

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To better distinguish this test from testResetOffsetsWithPartitionNoneLeader, maybe rename it to testResetOffsetsWithOfflinePartitionNotInResetTarget or something similar.

String topic = generateRandomTopic();
String group = "new.group";
String[] args = buildArgsForGroup(cluster, group, "--to-earliest", "--execute", "--topic", topic + ":1");
cluster.createTopic(topic, 3, (short) 2);

try (ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args);
Admin admin = cluster.admin()) {

@AndrewJSchofield AndrewJSchofield Jul 24, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: The indentation here is a bit strange not being a multiple of 4 spaces. I wonder whether just concatenating this on the end of the previous line would be simplest.

@TaiJuWu TaiJuWu Jul 24, 2025 •
edited
Loading

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

change to
try (Admin admin = cluster.admin(); ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args))

admin.alterPartitionReassignments(Map.of(new TopicPartition(topic, 2),
Optional.of(new NewPartitionReassignment(List.of(3, 4)))));

TestUtils.waitForCondition(() -> {
List<TopicPartitionInfo> partInfo = admin.describeTopics(List.of(topic)).topicNameValues().get(topic).get().partitions();
return partInfo.get(2).replicas().stream().map(Node::id).collect(Collectors.toUnmodifiableSet()).equals(Set.of(3, 4));
}, "Partitions did not complete reassignment to the target broker(s) within the expected time.");

cluster.shutdownBroker(3);
cluster.shutdownBroker(4);
TestUtils.waitForCondition(() -> !cluster.aliveBrokers().keySet().containsAll(Set.of(3, 4)),
"The cluster did not shut down the broker within the expected time.");

Map<TopicPartition, OffsetAndMetadata> resetOffsets = service.resetOffsets().get(group);
assertEquals(Set.of(new TopicPartition(topic, 1)), resetOffsets.keySet());
}
}

@ClusterTest
public void testResetOffsetsExistingTopic(ClusterInstance cluster) {
String topic = generateRandomTopic();
Expand Down
Loading