RamanVerma commented on code in PR #15384:
URL: https://github.com/apache/kafka/pull/15384#discussion_r1494984871
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala:
##########
@@ -278,12 +278,13 @@ class TransactionCoordinator(txnConfig: TransactionConfig,
def handleListTransactions(
filteredProducerIds: Set[Long],
- filteredStates: Set[String]
+ filteredStates: Set[String],
+ durationFilter: Long = -1
Review Comment:
@yyu1993 this default value needs to be changed to -1L as well
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java:
##########
@@ -61,6 +62,11 @@ public ListTransactionsOptions
filterProducerIds(Collection<Long> producerIdFilt
return this;
}
+ public ListTransactionsOptions durationFilter(long durationMs) {
Review Comment:
Please add a comment to this method like we have for the other methods
above. Also, we should probably change the method name to filterOnDuration, to
match rest of the filter setting methods.
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java:
##########
@@ -81,11 +87,16 @@ public Set<Long> filteredProducerIds() {
return filteredProducerIds;
}
+ public long getDurationFilter() {
Review Comment:
Please add a Java doc comment to the method. Also, change the method name to
`filteredDuration`
##########
clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java:
##########
@@ -81,11 +87,16 @@ public Set<Long> filteredProducerIds() {
return filteredProducerIds;
}
+ public long getDurationFilter() {
+ return durationFilter;
+ }
+
@Override
public String toString() {
return "ListTransactionsOptions(" +
"filteredStates=" + filteredStates +
", filteredProducerIds=" + filteredProducerIds +
+ ", durationFilter=" + durationFilter +
Review Comment:
nit: durationFilter -> filteredDuration
##########
tools/src/test/java/org/apache/kafka/tools/TransactionsCommandTest.java:
##########
@@ -187,14 +187,25 @@ private void testDescribeProducers(
assertEquals(expectedRows, new HashSet<>(table.subList(1,
table.size())));
}
- @Test
- public void testListTransactions() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testListTransactions(boolean hasDurationFilter) throws
Exception {
String[] args = new String[] {
"--bootstrap-server",
"localhost:9092",
"list"
};
+ if (hasDurationFilter) {
+ args = new String[] {
+ "--bootstrap-server",
+ "localhost:9092",
+ "list",
+ "--duration-filter",
+ Long.toString(Long.MAX_VALUE)
Review Comment:
hmm this will not return anything, right?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
