root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: Fix flaky tests in Tools modules (#20225) 

### Problem
The
`ShareGroupCommandTest.testDeleteShareGroupOffsetsArgsWithoutTopic()`,
`ShareGroupCommandTest.testDeleteShareGroupOffsetsArgsWithoutGroup()`,
`ResetStreamsGroupOffsetTest.testResetOffsetsWithoutGroupOption()`,
`DeleteStreamsGroupTest.testDeleteWithoutGroupOption()`,
`DescribeStreamsGroupTest.testDescribeWithoutGroupOption()` tests were
flaky due to a dependency on Set iteration order in error message
generation.

### Root Cause
The cleanup [commit](https://github.com/apache/kafka/pull/20091) that
replaced `new HashSet<>(Arrays.asList(...))` with `Set.of(...)` in
ShareGroupCommandOptions and StreamsGroupCommandOptions changed the
iteration characteristics of collections used for error message
generation:

This produces different orders like `[topic], [group]` vs `[group],
[topic]`, but the tests expected a specific order, causing intermittent
failures.

### Solution
Fix the root cause by ensuring deterministic error message generation
through alphabetical sorting of option names.

Reviewers: ShivsundarR <shr@confluent.io>, Ken Huang
 <s7133700@gmail.com>, TengYao Chi <frankvicky@apache.org>
This commit is contained in:
Sanskar Jhajharia 2025-07-23 12:10:18 +05:30 committed by GitHub
parent c7e4ff01cd
commit f1e9aa1c65
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 11 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandOptions.java
View File

@ -159,11 +159,11 @@ public class ShareGroupCommandOptions extends CommandDefaultOptions {
if (options.has(describeOpt)) {
if (!options.has(groupOpt) && !options.has(allGroupsOpt))
CommandLineUtils.printUsageAndExit(parser,
"Option " + describeOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + describeOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
List<OptionSpec<?>> mutuallyExclusiveOpts = List.of(membersOpt, offsetsOpt, stateOpt);
if (mutuallyExclusiveOpts.stream().mapToInt(o -> options.has(o) ? 1 : 0).sum() > 1) {
CommandLineUtils.printUsageAndExit(parser,
"Option " + describeOpt + " takes at most one of these options: " + mutuallyExclusiveOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + describeOpt + " takes at most one of these options: " + mutuallyExclusiveOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
}
if (options.has(stateOpt) && options.valueOf(stateOpt) != null)
CommandLineUtils.printUsageAndExit(parser,
@ -185,7 +185,7 @@ public class ShareGroupCommandOptions extends CommandDefaultOptions {
if (options.has(deleteOffsetsOpt)) {
if (!options.has(groupOpt) || !options.has(topicOpt))
CommandLineUtils.printUsageAndExit(parser,
"Option " + deleteOffsetsOpt + " takes the following options: " + allDeleteOffsetsOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + deleteOffsetsOpt + " takes the following options: " + allDeleteOffsetsOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
}
if (options.has(resetOffsetsOpt)) {

12
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommandOptions.java
View File

@ -222,7 +222,7 @@ public class StreamsGroupCommandOptions extends CommandDefaultOptions {
if (options.has(deleteOpt)) {
if (!options.has(groupOpt) && !options.has(allGroupsOpt))
CommandLineUtils.printUsageAndExit(parser,
"Option " + deleteOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + deleteOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
if (options.has(inputTopicOpt) || options.has(allInputTopicsOpt))
CommandLineUtils.printUsageAndExit(parser, "Kafka Streams does not support topic-specific offset " +
"deletion from a streams group.");
@ -253,11 +253,11 @@ public class StreamsGroupCommandOptions extends CommandDefaultOptions {
if (!options.has(groupOpt) && !options.has(allGroupsOpt))
CommandLineUtils.printUsageAndExit(parser,
"Option " + describeOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + describeOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
List<OptionSpec<?>> mutuallyExclusiveOpts = List.of(membersOpt, offsetsOpt, stateOpt);
if (mutuallyExclusiveOpts.stream().mapToInt(o -> options.has(o) ? 1 : 0).sum() > 1) {
CommandLineUtils.printUsageAndExit(parser,
"Option " + describeOpt + " takes at most one of these options: " + mutuallyExclusiveOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + describeOpt + " takes at most one of these options: " + mutuallyExclusiveOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
}
if (options.has(stateOpt) && options.valueOf(stateOpt) != null)
CommandLineUtils.printUsageAndExit(parser,
@ -267,7 +267,7 @@ public class StreamsGroupCommandOptions extends CommandDefaultOptions {
private void checkDeleteOffsetsArgs() {
if ((!options.has(inputTopicOpt) && !options.has(allInputTopicsOpt)) || !options.has(groupOpt))
CommandLineUtils.printUsageAndExit(parser,
"Option " + deleteOffsetsOpt + " takes the " + groupOpt + " and one of these options: " + allDeleteOffsetsOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + deleteOffsetsOpt + " takes the " + groupOpt + " and one of these options: " + allDeleteOffsetsOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
if (options.valuesOf(groupOpt).size() > 1)
CommandLineUtils.printUsageAndExit(parser,
"Option " + deleteOffsetsOpt + " supports only one " + groupOpt + " at a time, but found: " + options.valuesOf(groupOpt));
@ -286,7 +286,7 @@ public class StreamsGroupCommandOptions extends CommandDefaultOptions {
if (!options.has(groupOpt) && !options.has(allGroupsOpt))
CommandLineUtils.printUsageAndExit(parser,
"Option " + resetOffsetsOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + resetOffsetsOpt + " takes one of these options: " + allGroupSelectionScopeOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
CommandLineUtils.checkInvalidArgs(parser, options, resetToOffsetOpt, minus(allResetOffsetScenarioOpts, resetToOffsetOpt));
CommandLineUtils.checkInvalidArgs(parser, options, resetToDatetimeOpt, minus(allResetOffsetScenarioOpts, resetToDatetimeOpt));
@ -301,7 +301,7 @@ public class StreamsGroupCommandOptions extends CommandDefaultOptions {
private void checkDeleteAllInternalTopicsArgs() {
if (!options.has(resetOffsetsOpt) && !options.has(deleteOpt)) {
CommandLineUtils.printUsageAndExit(parser,
"Option " + deleteAllInternalTopicsOpt + " takes one of these options: " + allDeleteInternalGroupsOpts.stream().map(Object::toString).collect(Collectors.joining(", ")));
"Option " + deleteAllInternalTopicsOpt + " takes one of these options: " + allDeleteInternalGroupsOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(", ")));
} else if (options.has(resetOffsetsOpt) && !options.has(executeOpt)) {
CommandLineUtils.printUsageAndExit(parser,
"Option " + deleteAllInternalTopicsOpt + " takes " + executeOpt + " when " + resetOffsetsOpt + " is used.");

4
tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java
View File

@ -626,7 +626,7 @@ public class ShareGroupCommandTest {
AtomicBoolean exited = new AtomicBoolean(false);
Exit.setExitProcedure(((statusCode, message) -> {
assertNotEquals(0, statusCode);
assertTrue(message.contains("Option [delete-offsets] takes the following options: [topic], [group]"));
assertTrue(message.contains("Option [delete-offsets] takes the following options: [group], [topic]"));
exited.set(true);
}));
try {
@ -646,7 +646,7 @@ public class ShareGroupCommandTest {
AtomicBoolean exited = new AtomicBoolean(false);
Exit.setExitProcedure(((statusCode, message) -> {
assertNotEquals(0, statusCode);
assertTrue(message.contains("Option [delete-offsets] takes the following options: [topic], [group]"));
assertTrue(message.contains("Option [delete-offsets] takes the following options: [group], [topic]"));
exited.set(true);
}));
try {