Copilot commented on code in PR #22786:
URL: https://github.com/apache/kafka/pull/22786#discussion_r3541812744
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -602,6 +665,14 @@ public static VerifiableShareConsumer
createFromArgs(ArgumentParser parser, Stri
String configFile = res.getString("commandConfig");
String brokerHostAndPort = res.getString("bootstrapServer");
+ List<AcknowledgeType> ackPattern =
parseAckPattern(res.getString("ackPattern"));
+ if (!ackPattern.isEmpty()) {
+ if (acknowledgementMode == AcknowledgementMode.AUTO) {
+ throw new ArgumentParserException(
+ "--ack-pattern requires --acknowledgement-mode to be
'sync' or 'async', not 'auto'", parser);
+ }
+ }
Review Comment:
`--ack-pattern` parsing can throw `IllegalArgumentException` (e.g., typo
like "accep" or a trailing comma), which will bypass the existing
`ArgumentParserException` handling and crash with a stack trace instead of a
clean CLI error. Catch parse failures and rethrow as `ArgumentParserException`
with a helpful message.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -410,10 +432,27 @@ private void onRecordsReceived(ConsumerRecords<String,
String> records) {
public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap,
Exception exception) {
List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
int totalAcknowledged = 0;
+ Map<String, Long> ackTypeCounts = ackPattern.isEmpty() ? null : new
HashMap<>();
for (Map.Entry<TopicIdPartition, Set<Long>> offsetEntry :
offsetsMap.entrySet()) {
Review Comment:
`pendingAckTypes` entries are removed whenever the acknowledgement commit
callback fires, even when the callback has a non-null exception. If the client
retries the same acknowledgement batch internally (without redelivery), later
successful callbacks won't be able to attribute ack types, so `ackTypeCounts`
can be underreported/flaky. Only generate `ackTypeCounts` (and remove from
`pendingAckTypes`) on successful callbacks.
##########
tests/kafkatest/services/kafka/kafka.py:
##########
@@ -907,6 +917,12 @@ def start_node(self, node, timeout_sec=60, **kwargs):
else:
if get_version(node).supports_feature_command():
cmd += " --feature transaction.version=0"
+ if self.share_version is not None:
+ # share.version=2 (KIP-1191 DLQ) declares a bootstrap
metadata.version of 4.4-IV0,
+ # but that isn't yet a valid --release-version for
kafka-storage.sh format in this
+ # build (max supported is 4.3-IV0), and
Feature.validateVersion() does not enforce
+ # the dependency, so the feature can be bootstrapped on its
own.
+ cmd += " --feature share.version=%s" % self.share_version
Review Comment:
When `share_version` is set, `kafka-storage.sh format` is always given a
`--feature share.version=...` flag, even on Kafka versions that don't support
`--feature` formatting. This can break upgrade/mixed-version ducktape runs.
Consider guarding with `get_version(node).supports_feature_command()` and
failing fast with a clear error if unsupported.
--
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]
