Address pr's comment and rebase

apache · Oct 21, 2022 · f3f9c78 · f3f9c78
1 parent f87603f
commit f3f9c78
Show file tree

Hide file tree

Showing 5 changed files with 35 additions and 16 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/TransactionsBase.java
@@ -78,24 +78,36 @@ protected void internalListCoordinators(AsyncResponse asyncResponse) {
             asyncResponse.resume(new RestException(ex));
             return;
         }
+        Map<Integer, TransactionCoordinatorInfo> result = new HashMap<>();
         admin.lookups()
                 .lookupPartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.getPartitionedTopicName())
-                .thenAccept(map -> {
-                    if (map.isEmpty()) {
-                        asyncResponse.resume(new RestException(Response.Status.NOT_FOUND,
-                                "Transaction coordinator not found"));
-                        return;
-                    }
-                    Map<Integer, TransactionCoordinatorInfo> result = new HashMap<>();
+                .thenCompose(map -> {
                     map.forEach((topicPartition, brokerServiceUrl) -> {
                         final int coordinatorId = TopicName.getPartitionIndex(topicPartition);
-                        result.put(coordinatorId, new TransactionCoordinatorInfo(brokerServiceUrl));
+                        result.put(coordinatorId, new TransactionCoordinatorInfo(coordinatorId, brokerServiceUrl));
                     });
-                    asyncResponse.resume(result);
 
+                    return getPulsarResources()
+                            .getTopicResources()
+                            .getExistingPartitions(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN);
+                })
+                .thenAccept(allPartitions -> {
+                    allPartitions
+                            .stream()
+                            .filter(partition -> SystemTopicNames
+                                    .isTransactionCoordinatorAssign(TopicName.get(partition)))
+                            .forEach(partition -> {
+                                final int coordinatorId = TopicName.getPartitionIndex(partition);
+                                if (!result.containsKey(coordinatorId)) {
+                                    result.put(coordinatorId,
+                                            new TransactionCoordinatorInfo(coordinatorId, null));
+                                }
+                            });
+                    asyncResponse.resume(result.values());
                 })
                 .exceptionally(ex -> {
-                    log.error("[{}] Failed to list transaction coordinators.", clientAppId(), ex);
+                    log.error("[{}] Failed to list transaction coordinators: {}",
+                            clientAppId(), ex.getMessage(), ex);
                     resumeAsyncResponseExceptionally(asyncResponse, ex);
                     return null;
                 });

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java
@@ -113,8 +113,9 @@ protected void cleanup() throws Exception {
     @Test(timeOut = 20000)
     public void testListTransactionCoordinators() throws Exception {
         initTransaction(4);
-        final Map<Integer, TransactionCoordinatorInfo> result = admin
+        final List<TransactionCoordinatorInfo> result = admin
                 .transactions().listTransactionCoordinatorsAsync().get();
+        System.out.println("result" + result);
         assertEquals(result.size(), 4);
         final String expectedUrl = pulsar.getBrokerServiceUrl();
         for (int i = 0; i < 4; i++) {

diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Transactions.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Transactions.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.admin;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -40,14 +41,14 @@ public interface Transactions {
      *
      * @return the transaction coordinators list.
      */
-    Map<Integer, TransactionCoordinatorInfo> listTransactionCoordinators() throws PulsarAdminException;
+    List<TransactionCoordinatorInfo> listTransactionCoordinators() throws PulsarAdminException;
 
     /**
      * List transaction coordinators.
      *
      * @return the future of the transaction coordinators list.
      */
-    CompletableFuture<Map<Integer, TransactionCoordinatorInfo>> listTransactionCoordinatorsAsync();
+    CompletableFuture<List<TransactionCoordinatorInfo>> listTransactionCoordinatorsAsync();
 
 
     /**

diff --git a/...-api/src/main/java/org/apache/pulsar/common/policies/data/TransactionCoordinatorInfo.java b/...-api/src/main/java/org/apache/pulsar/common/policies/data/TransactionCoordinatorInfo.java
@@ -22,13 +22,16 @@
 import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import lombok.ToString;
 
 /**
  * Transaction coordinator information.
  */
 @Getter
 @NoArgsConstructor
 @AllArgsConstructor
+@ToString
 public class TransactionCoordinatorInfo {
+    private long id;
     private String brokerServiceUrl;
 }
diff --git a/...-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TransactionsImpl.java b/...-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TransactionsImpl.java
@@ -19,6 +19,8 @@
 package org.apache.pulsar.client.admin.internal;
 
 import static com.google.common.base.Preconditions.checkArgument;
+
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -50,14 +52,14 @@ public TransactionsImpl(WebTarget web, Authentication auth, long readTimeoutMs)
     }
 
     @Override
-    public Map<Integer, TransactionCoordinatorInfo> listTransactionCoordinators() throws PulsarAdminException {
+    public List<TransactionCoordinatorInfo> listTransactionCoordinators() throws PulsarAdminException {
         return sync(() -> listTransactionCoordinatorsAsync());
     }
 
     @Override
-    public CompletableFuture<Map<Integer, TransactionCoordinatorInfo>> listTransactionCoordinatorsAsync() {
+    public CompletableFuture<List<TransactionCoordinatorInfo>> listTransactionCoordinatorsAsync() {
         WebTarget path = adminV3Transactions.path("coordinators");
-        return asyncGetRequest(path, new FutureCallback<Map<Integer, TransactionCoordinatorInfo>>(){});
+        return asyncGetRequest(path, new FutureCallback<List<TransactionCoordinatorInfo>>(){});
     }
 
     @Override