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 16 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
3 changes: 2 additions & 1 deletion 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 @@ -655,6 +655,7 @@ private List<TopicPartition> filterNoneLeaderPartitions(Collection<TopicPartitio
.flatMap(entry -> entry.getValue().partitions().stream()
.filter(partitionInfo -> partitionInfo.leader() == null)
.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))

.filter(topicPartitions::contains)
.toList();
} catch (Exception e) {
throw new RuntimeException(e);
Expand Down Expand Up @@ -1054,7 +1055,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
25 changes: 25 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 @@ -27,6 +27,7 @@
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.LeaderNotAvailableException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.serialization.ByteArraySerializer;
Expand All @@ -52,6 +53,7 @@
import java.util.Map;
import java.util.Objects;
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 +155,29 @@ public void testResetOffsetsNotExistingGroup(ClusterInstance cluster) throws Exc
}
}

@ClusterTest(
brokers = 2,
serverProperties = {
@ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"),
@ClusterConfigProperty(key = TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, value = "1")

@Yunyung Yunyung Jul 25, 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.

Do we really need to set these properties? We can reproduce the failure without setting them.

@TaiJuWu TaiJuWu Jul 25, 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.

These settings can avoid flaky.
In my mind, if __consumer_offset is at broker 1 and we shutdown it, we don't have any __consumer_offset.

@m1a2st m1a2st Jul 30, 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.

The min.insync.replicas default value is 1, so we don't need to configure it explicitly?

}
)
public void testResetOffsetsWithOfflinePartitionNotInResetTarget(ClusterInstance cluster) throws Exception {
String topic = generateRandomTopic();
String group = "new.group";
String[] args = buildArgsForGroup(cluster, group, "--to-earliest", "--execute", "--topic", topic + ":0");

try (Admin admin = cluster.admin(); ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
admin.createTopics(List.of(new NewTopic(topic, Map.of(0, List.of(0), 1, List.of(1)))));
cluster.waitTopicCreation(topic, 2);

cluster.shutdownBroker(1);

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

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