KAFKA-19073: add transactional ID pattern filter to ListTransactions #19355
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
github-actions
bot
added
triage
CalvinConfluent
changed the title
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
chia7712
reviewed
Apr 15, 2025
| @@ -343,6 +345,9 @@ class TransactionStateManager(brokerId: Int, | |||
| false | |||
| } else if (filterDurationMs >= 0 && (now - txnMetadata.txnStartTimestamp) <= filterDurationMs) { | |||
| false | |||
| } else if (!filterTransactionalIdPattern.isEmpty && | |||
There was a problem hiding this comment.
filterTransactionalIdPattern.nonEmpty
| @@ -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", | |||
There was a problem hiding this comment.
Could you please add version check to ListTransactionsRequest.Builder#build?
if (data.transactionalIdPatternFilter() != null && version < 2) {
throw new UnsupportedVersionException("xxxx");
}| @@ -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()) { | |||
There was a problem hiding this comment.
Should we avoid compiling the regex for each transaction id?
| @@ -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, | |||
There was a problem hiding this comment.
The protocol says if null, all transactions are returned; Should we align it with null?
BTW, ConsumerGroupHeartbeatRequest uses null
m1a2st
reviewed
Apr 16, 2025
| 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()); |
There was a problem hiding this comment.
Collections.emptyList() can instead of List.of()
| ListTransactionsOptions options = new ListTransactionsOptions() | ||
| .filterOnTransactionalIdPattern(filteredTransactionalIdPattern); | ||
| ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext); | ||
| ListTransactionsRequest request = handler.buildBatchedRequest(brokerId, singleton(brokerKey)).build(); |
There was a problem hiding this comment.
singleton() can instead of Set.of()
Comment on lines
+318
to
+323
| transactions.put(0, asList( | ||
| new TransactionListing("bar", 98765L, TransactionState.PREPARE_ABORT) | ||
| )); | ||
| transactions.put(1, singletonList( | ||
| new TransactionListing("baz", 13579L, TransactionState.COMPLETE_COMMIT) | ||
| )); |
There was a problem hiding this comment.
asList() and singletonList can instead of List.of()
| List<String> expectedHeaders = TransactionsCommand.ListTransactionsCommand.HEADERS; | ||
| assertEquals(expectedHeaders, table.get(0)); | ||
| Set<List<String>> expectedRows = Set.of( | ||
| asList("bar", "0", "98765", "PrepareAbort"), |
There was a problem hiding this comment.
asList() can instead of List.of()
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
https://issues.apache.org/jira/browse/KAFKA-19073