Make the pattern nullable

Author: CalvinConfluent

Diff for: clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java

@@ -75,7 +75,9 @@ public ListTransactionsRequest.Builder buildBatchedRequest(
             .map(TransactionState::toString)
             .collect(Collectors.toList()));
         request.setDurationFilter(options.filteredDuration());
-        request.setTransactionalIdPatternFilter(options.filteredTransactionalIdPattern());
+        if (!options.filteredTransactionalIdPattern().isEmpty()) {
+            request.setTransactionalIdPatternFilter(options.filteredTransactionalIdPattern());
+        }
         return new ListTransactionsRequest.Builder(request);
     }
 

Diff for: clients/src/main/resources/common/message/ListTransactionsRequest.json

@@ -33,8 +33,8 @@
     { "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+",
-      "about": "The transactional ID regular expression pattern to filter by: if empty, all transactions are returned; If non-empty, then only the transactions matching with the pattern will be returned."
+    { "name": "TransactionalIdPatternFilter", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
+      "about": "The transactional ID regular expression pattern to filter by: if null, all transactions are returned; otherwise, then only the transactions matching with the pattern will be returned."
     }
   ]
 }

Diff for: clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java

@@ -43,6 +43,7 @@
 import static java.util.Collections.singleton;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class ListTransactionsHandlerTest {
@@ -100,6 +101,18 @@ public void testBuildRequestWithFilteredTransactionalIdPattern() {
         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();
+        assertNull(request.data().transactionalIdPatternFilter());
+    }
+
     @Test
     public void testBuildRequestWithDurationFilter() {
         int brokerId = 1;

Diff for: core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala

@@ -317,7 +317,7 @@ class TransactionStateManager(brokerId: Int,
     filterProducerIds: Set[Long],
     filterStateNames: Set[String],
     filterDurationMs: Long,
-    filteredTransactionalIdPattern: String
+    filterTransactionalIdPattern: String
   ): ListTransactionsResponseData = {
     inReadLock(stateLock) {
       val response = new ListTransactionsResponseData()
@@ -345,8 +345,8 @@ class TransactionStateManager(brokerId: Int,
             false
           } else if (filterDurationMs >= 0 && (now - txnMetadata.txnStartTimestamp) <= filterDurationMs) {
             false
-          } else if (!filteredTransactionalIdPattern.isEmpty &&
-            !Pattern.compile(filteredTransactionalIdPattern).matcher(txnMetadata.transactionalId).matches()) {
+          } else if (!filterTransactionalIdPattern.isEmpty &&
+            !Pattern.compile(filterTransactionalIdPattern).matcher(txnMetadata.transactionalId).matches()) {
             false
           } else {
             true

Diff for: core/src/main/scala/kafka/server/KafkaApis.scala

@@ -2472,7 +2472,11 @@ 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 transactionalIdPatternFilter = listTransactionsRequest.data.transactionalIdPatternFilter
+    val transactionalIdPatternFilter = if (listTransactionsRequest.data.transactionalIdPatternFilter == null) {
+      ""
+    } else {
+      listTransactionsRequest.data.transactionalIdPatternFilter
+    }
     val response = txnCoordinator.handleListTransactions(
       filteredProducerIds,
       filteredStates,