You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "CalvinConfluent (via GitHub)" <gi...@apache.org> on 2025/04/02 22:38:10 UTC
[PR] KAFKA-19073: add transactional ID prefix filter to ListTransactions [kafka]
CalvinConfluent opened a new pull request, #19355:
URL: https://github.com/apache/kafka/pull/19355
https://issues.apache.org/jira/browse/KAFKA-19073
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2056908883
##########
clients/src/main/resources/common/message/ListTransactionsRequest.json:
##########
@@ -30,6 +32,9 @@
},
{ "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
"about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned."
+ },
+ { "name": "TransactionalIdPatternFilter", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
Review Comment:
Sure, added.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2067459157
##########
clients/src/main/resources/common/message/ListTransactionsRequest.json:
##########
@@ -30,6 +32,9 @@
},
{ "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
"about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned."
+ },
+ { "name": "TransactionalIdPattern", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
Review Comment:
Any reason why we can't just accept null on the client side? We convert to "" now.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "brandboat (via GitHub)" <gi...@apache.org>.
brandboat commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2066884073
##########
clients/src/main/java/org/apache/kafka/common/requests/ListTransactionsRequest.java:
##########
@@ -38,6 +38,10 @@ public ListTransactionsRequest build(short version) {
throw new UnsupportedVersionException("Duration filter can be set only when using API version 1 or higher." +
" If client is connected to an older broker, do not specify duration filter or set duration filter to -1.");
}
+ if (data.transactionalIdPattern() != null && version < 2) {
+ throw new UnsupportedVersionException("Transactional ID pattern filter can be set only when using API version 1 or higher." +
+ " If client is connected to an older broker, do not specify the pattern filter.");
Review Comment:
Not sure if I misunderstood anything, but Transactional ID pattern is introduced in api version 2+ right?
Doe that mean the error message should be `...using API version 2 or higher.` ?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2068904148
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2472,7 +2472,17 @@ class KafkaApis(val requestChannel: RequestChannel,
val filteredProducerIds = listTransactionsRequest.data.producerIdFilters.asScala.map(Long.unbox).toSet
val filteredStates = listTransactionsRequest.data.stateFilters.asScala.toSet
val durationFilter = listTransactionsRequest.data.durationFilter()
- val response = txnCoordinator.handleListTransactions(filteredProducerIds, filteredStates, durationFilter)
+ val transactionalIdPatternFilter = if (listTransactionsRequest.data.transactionalIdPattern == null) {
+ ""
Review Comment:
I see. I was looking at the lines directly above where we do check for null. It's a bit confusing which things check for null vs empty 😓
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2056904119
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -343,6 +345,9 @@ class TransactionStateManager(brokerId: Int,
false
} else if (filterDurationMs >= 0 && (now - txnMetadata.txnStartTimestamp) <= filterDurationMs) {
false
+ } else if (!filterTransactionalIdPattern.isEmpty &&
+ !Pattern.compile(filterTransactionalIdPattern).matcher(txnMetadata.transactionalId).matches()) {
Review Comment:
Sounds good.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java:
##########
@@ -70,6 +71,19 @@ public ListTransactionsOptions filterOnDuration(long durationMs) {
return this;
}
+ /**
+ * Filter only the transactions that match with the given transactional ID pattern.
+ * If no filter is specified or if the passed string is empty,
Review Comment:
Good catch! Thanks
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2045050285
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -343,6 +345,9 @@ class TransactionStateManager(brokerId: Int,
false
} else if (filterDurationMs >= 0 && (now - txnMetadata.txnStartTimestamp) <= filterDurationMs) {
false
+ } else if (!filterTransactionalIdPattern.isEmpty &&
Review Comment:
`filterTransactionalIdPattern.nonEmpty`
##########
clients/src/main/resources/common/message/ListTransactionsRequest.json:
##########
@@ -30,6 +32,9 @@
},
{ "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
"about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned."
+ },
+ { "name": "TransactionalIdPatternFilter", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
Review Comment:
Could you please add version check to `ListTransactionsRequest.Builder#build`?
```java
if (data.transactionalIdPatternFilter() != null && version < 2) {
throw new UnsupportedVersionException("xxxx");
}
```
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java:
##########
@@ -70,6 +71,19 @@ public ListTransactionsOptions filterOnDuration(long durationMs) {
return this;
}
+ /**
+ * Filter only the transactions that match with the given transactional ID pattern.
+ * If no filter is specified or if the passed string is empty,
Review Comment:
The protocol says `if null, all transactions are returned;` Should we align it with `null`?
BTW, `ConsumerGroupHeartbeatRequest` uses `null`
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -343,6 +345,9 @@ class TransactionStateManager(brokerId: Int,
false
} else if (filterDurationMs >= 0 && (now - txnMetadata.txnStartTimestamp) <= filterDurationMs) {
false
+ } else if (!filterTransactionalIdPattern.isEmpty &&
+ !Pattern.compile(filterTransactionalIdPattern).matcher(txnMetadata.transactionalId).matches()) {
Review Comment:
Should we avoid compiling the regex for each transaction id?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on PR #19355:
URL: https://github.com/apache/kafka/pull/19355#issuecomment-2845169525
@chia7712 Any other comments? Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2066815028
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java:
##########
@@ -75,6 +75,9 @@ public ListTransactionsRequest.Builder buildBatchedRequest(
.map(TransactionState::toString)
.collect(Collectors.toList()));
request.setDurationFilter(options.filteredDuration());
+ if (!options.filteredTransactionalIdPattern().isEmpty()) {
Review Comment:
For the filteredTransactionalIdPattern in the ListTransactionsOptions, the default value is "" and we only update it if the user sets the `--transactional-id-pattern`. It should be valid.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2067158894
##########
clients/src/main/resources/common/message/ListTransactionsRequest.json:
##########
@@ -30,6 +32,9 @@
},
{ "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
"about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned."
+ },
+ { "name": "TransactionalIdPattern", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
Review Comment:
I'm a little confused by why we convert null to "" on the client side if we allow null and it's the same behavior as ""
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2068724036
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2472,7 +2472,17 @@ class KafkaApis(val requestChannel: RequestChannel,
val filteredProducerIds = listTransactionsRequest.data.producerIdFilters.asScala.map(Long.unbox).toSet
val filteredStates = listTransactionsRequest.data.stateFilters.asScala.toSet
val durationFilter = listTransactionsRequest.data.durationFilter()
- val response = txnCoordinator.handleListTransactions(filteredProducerIds, filteredStates, durationFilter)
+ val transactionalIdPatternFilter = if (listTransactionsRequest.data.transactionalIdPattern == null) {
+ ""
Review Comment:
```scala
val pattern = if (filterTransactionalIdPattern.nonEmpty) {
try {
Pattern.compile(filterTransactionalIdPattern)
}
catch {
case e: PatternSyntaxException =>
throw new InvalidRegularExpression(String.format("Transaction ID pattern `%s` is not a valid regular expression: %s.", filterTransactionalIdPattern, e.getMessage))
}
} else null
```
It seems `TransactionStateManager` does not check the null for `filterTransactionalIdPattern`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on PR #19355:
URL: https://github.com/apache/kafka/pull/19355#issuecomment-2835689257
@chia7712 @m1a2st The KIP-1152 has been accepted! Can you help take another look? Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2067455423
##########
clients/src/main/resources/common/message/ListTransactionsRequest.json:
##########
@@ -30,6 +32,9 @@
},
{ "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
"about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned."
+ },
+ { "name": "TransactionalIdPattern", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
Review Comment:
Yeah, I think it is because the other regex fields allow null. So we want to get they aligned, like `SubscribedTopicRegex` in ConsumerGroupHeartbeatRequest. As for converting to an empty string, I find it is convenient to handle in code.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2066833385
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java:
##########
@@ -75,6 +75,9 @@ public ListTransactionsRequest.Builder buildBatchedRequest(
.map(TransactionState::toString)
.collect(Collectors.toList()));
request.setDurationFilter(options.filteredDuration());
+ if (!options.filteredTransactionalIdPattern().isEmpty()) {
Review Comment:
or could you please add null check to `ListTransactionsOptions#filterOnTransactionalIdPattern` if we don't want to user to set it to null.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2066907375
##########
clients/src/main/java/org/apache/kafka/common/requests/ListTransactionsRequest.java:
##########
@@ -38,6 +38,10 @@ public ListTransactionsRequest build(short version) {
throw new UnsupportedVersionException("Duration filter can be set only when using API version 1 or higher." +
" If client is connected to an older broker, do not specify duration filter or set duration filter to -1.");
}
+ if (data.transactionalIdPattern() != null && version < 2) {
+ throw new UnsupportedVersionException("Transactional ID pattern filter can be set only when using API version 1 or higher." +
+ " If client is connected to an older broker, do not specify the pattern filter.");
Review Comment:
Thanks! Good catch! Updated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2066869605
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java:
##########
@@ -75,6 +75,9 @@ public ListTransactionsRequest.Builder buildBatchedRequest(
.map(TransactionState::toString)
.collect(Collectors.toList()));
request.setDurationFilter(options.filteredDuration());
+ if (!options.filteredTransactionalIdPattern().isEmpty()) {
Review Comment:
Sounds good, updated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2067601848
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2472,7 +2472,17 @@ class KafkaApis(val requestChannel: RequestChannel,
val filteredProducerIds = listTransactionsRequest.data.producerIdFilters.asScala.map(Long.unbox).toSet
val filteredStates = listTransactionsRequest.data.stateFilters.asScala.toSet
val durationFilter = listTransactionsRequest.data.durationFilter()
- val response = txnCoordinator.handleListTransactions(filteredProducerIds, filteredStates, durationFilter)
+ val transactionalIdPatternFilter = if (listTransactionsRequest.data.transactionalIdPattern == null) {
+ ""
Review Comment:
Do we need this as well? We have a null check in TransactionStateManager
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #19355:
URL: https://github.com/apache/kafka/pull/19355
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19355:
URL: https://github.com/apache/kafka/pull/19355#issuecomment-2791447052
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2045837695
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java:
##########
@@ -86,6 +87,32 @@ public void testBuildRequestWithFilteredState() {
assertEquals(Collections.emptyList(), request.data().producerIdFilters());
}
+
+ @Test
+ public void testBuildRequestWithFilteredTransactionalIdPattern() {
+ int brokerId = 1;
+ BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId));
+ String filteredTransactionalIdPattern = "^special-.*";
+ ListTransactionsOptions options = new ListTransactionsOptions()
+ .filterOnTransactionalIdPattern(filteredTransactionalIdPattern);
+ ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext);
+ ListTransactionsRequest request = handler.buildBatchedRequest(brokerId, singleton(brokerKey)).build();
+ assertEquals(filteredTransactionalIdPattern, request.data().transactionalIdPatternFilter());
+ assertEquals(Collections.emptyList(), request.data().stateFilters());
Review Comment:
`Collections.emptyList()` can instead of `List.of()`
##########
clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java:
##########
@@ -86,6 +87,32 @@ public void testBuildRequestWithFilteredState() {
assertEquals(Collections.emptyList(), request.data().producerIdFilters());
}
+
+ @Test
+ public void testBuildRequestWithFilteredTransactionalIdPattern() {
+ int brokerId = 1;
+ BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId));
+ String filteredTransactionalIdPattern = "^special-.*";
+ ListTransactionsOptions options = new ListTransactionsOptions()
+ .filterOnTransactionalIdPattern(filteredTransactionalIdPattern);
+ ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext);
+ ListTransactionsRequest request = handler.buildBatchedRequest(brokerId, singleton(brokerKey)).build();
+ assertEquals(filteredTransactionalIdPattern, request.data().transactionalIdPatternFilter());
+ assertEquals(Collections.emptyList(), request.data().stateFilters());
+ }
+
+ @Test
+ public void testBuildRequestWithEmptyFilteredTransactionalIdPattern() {
+ int brokerId = 1;
+ BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId));
+ String filteredTransactionalIdPattern = "";
+ ListTransactionsOptions options = new ListTransactionsOptions()
+ .filterOnTransactionalIdPattern(filteredTransactionalIdPattern);
+ ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext);
+ ListTransactionsRequest request = handler.buildBatchedRequest(brokerId, singleton(brokerKey)).build();
Review Comment:
`singleton()` can instead of `Set.of()`
##########
tools/src/test/java/org/apache/kafka/tools/TransactionsCommandTest.java:
##########
@@ -304,6 +304,42 @@ public void testDescribeTransaction() throws Exception {
assertEquals(expectedRow, table.get(1));
}
+ @Test
+ public void testListTransactionsWithTransactionalIdPattern() throws Exception {
+ String[] args = new String[] {
+ "--bootstrap-server",
+ "localhost:9092",
+ "list",
+ "--transactional-id-pattern-filter",
+ "ba.*"
+ };
+
+ Map<Integer, Collection<TransactionListing>> transactions = new HashMap<>();
+ transactions.put(0, asList(
+ new TransactionListing("bar", 98765L, TransactionState.PREPARE_ABORT)
+ ));
+ transactions.put(1, singletonList(
+ new TransactionListing("baz", 13579L, TransactionState.COMPLETE_COMMIT)
+ ));
+
+ expectListTransactions(new ListTransactionsOptions().filterOnTransactionalIdPattern("ba.*"), transactions);
+
+ execute(args);
+ assertNormalExit();
+
+ List<List<String>> table = readOutputAsTable();
+ assertEquals(3, table.size());
+
+ // Assert expected headers
+ List<String> expectedHeaders = TransactionsCommand.ListTransactionsCommand.HEADERS;
+ assertEquals(expectedHeaders, table.get(0));
+ Set<List<String>> expectedRows = Set.of(
+ asList("bar", "0", "98765", "PrepareAbort"),
Review Comment:
`asList()` can instead of `List.of()`
##########
tools/src/test/java/org/apache/kafka/tools/TransactionsCommandTest.java:
##########
@@ -304,6 +304,42 @@ public void testDescribeTransaction() throws Exception {
assertEquals(expectedRow, table.get(1));
}
+ @Test
+ public void testListTransactionsWithTransactionalIdPattern() throws Exception {
+ String[] args = new String[] {
+ "--bootstrap-server",
+ "localhost:9092",
+ "list",
+ "--transactional-id-pattern-filter",
+ "ba.*"
+ };
+
+ Map<Integer, Collection<TransactionListing>> transactions = new HashMap<>();
+ transactions.put(0, asList(
+ new TransactionListing("bar", 98765L, TransactionState.PREPARE_ABORT)
+ ));
+ transactions.put(1, singletonList(
+ new TransactionListing("baz", 13579L, TransactionState.COMPLETE_COMMIT)
+ ));
Review Comment:
`asList()` and `singletonList` can instead of `List.of()`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2067158894
##########
clients/src/main/resources/common/message/ListTransactionsRequest.json:
##########
@@ -30,6 +32,9 @@
},
{ "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
"about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned."
+ },
+ { "name": "TransactionalIdPattern", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
Review Comment:
I'm a little confused by why we convert null to "" on the client side if we allow null and it's the same result as "" (but we do the extra step of filtering for "" on the server side)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2067560540
##########
clients/src/main/resources/common/message/ListTransactionsRequest.json:
##########
@@ -30,6 +32,9 @@
},
{ "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
"about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned."
+ },
+ { "name": "TransactionalIdPattern", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
Review Comment:
I have updated the client side not to convert null in `ListTransactionsOptions#filterOnTransactionalIdPattern`. Also adding necessary null handlings.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2066607585
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java:
##########
@@ -75,6 +75,9 @@ public ListTransactionsRequest.Builder buildBatchedRequest(
.map(TransactionState::toString)
.collect(Collectors.toList()));
request.setDurationFilter(options.filteredDuration());
+ if (!options.filteredTransactionalIdPattern().isEmpty()) {
Review Comment:
`filteredTransactionalIdPattern` is nullable, so should we add null check?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19073: add transactional ID pattern filter to ListTransactions [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #19355:
URL: https://github.com/apache/kafka/pull/19355#discussion_r2068961294
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2472,7 +2472,17 @@ class KafkaApis(val requestChannel: RequestChannel,
val filteredProducerIds = listTransactionsRequest.data.producerIdFilters.asScala.map(Long.unbox).toSet
val filteredStates = listTransactionsRequest.data.stateFilters.asScala.toSet
val durationFilter = listTransactionsRequest.data.durationFilter()
- val response = txnCoordinator.handleListTransactions(filteredProducerIds, filteredStates, durationFilter)
+ val transactionalIdPatternFilter = if (listTransactionsRequest.data.transactionalIdPattern == null) {
+ ""
Review Comment:
The main reason to convert null to an empty string is to avoid null handling in the downstream. But yeah, looks like there is only one place to check null. Updated to use null in the code path.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org