aliehsaeedii commented on code in PR #21826:
URL: https://github.com/apache/kafka/pull/21826#discussion_r3813311533
##########
tools/src/test/java/org/apache/kafka/tools/ResetIntegrationTest.java:
##########
@@ -268,6 +271,7 @@ public void
shouldNotAllowToResetWhenIntermediateTopicAbsent(ClusterInstance clu
final String appId = generateAppId();
final String[] parameters = new String[] {
"--application-id", appId,
+ "--force",
Review Comment:
`shouldNotAllowToResetWhenInputTopicAbsent` above still has no `--force`.
Its app id has no consumer group, so the new check exits 1 there before the
missing-input-topic path runs — it now passes for the wrong reason. Add
`--force` like you did for the other negative tests.
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -521,6 +562,7 @@ private Map<TopicPartition, Long> checkOffsetRange(final
Map<TopicPartition, Lon
}
private int maybeDeleteInternalTopics(final Admin adminClient, final
StreamsResetterOptions options) {
+
Review Comment:
Stray blank line — drop it to match the other methods.
##########
tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java:
##########
@@ -431,4 +432,111 @@ public synchronized Map<TopicPartition,
OffsetAndTimestamp> offsetsForTimes(fina
return topicPartitionToOffsetAndTimestamp;
}
}
+
+ @Test
+ public void shouldFailIfApplicationIdDoesNotExistAsConsumerGroup() throws
Exception {
Review Comment:
This hits the same not-found branch as `shouldFailWithTypoInApplicationId`,
which also asserts the message. Consider dropping this one and keeping the typo
test.
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -181,6 +190,38 @@ public int execute(final String[] args, final Properties
config) {
}
}
+ // Note: this check confirms the application.id exists as a consumer group,
+ // but cannot prevent topic deletion for other applications whose IDs share
+ // this application.id as a prefix (e.g. resetting "foo" may affect
"foo-v2"
+ // topics). This is a known limitation of prefix-based topic inference.
+ void validateApplicationIdExists(final String applicationId,
+ final Admin adminClient)
+ throws InterruptedException, TimeoutException {
+ ConsumerGroupDescription description = null;
+ try {
+ final Map<String, ConsumerGroupDescription> groups = adminClient
+ .describeConsumerGroups(Set.of(applicationId))
Review Comment:
The PR description says this calls `listGroups()`, but it uses
`describeConsumerGroups` — update the description before merge.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
