[FEATURE] Add KopEventManager to repair metadata consistency of cluster and consistency of consumer group status

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AdminManager.java

@@ -21,16 +21,20 @@
 import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation;
 import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationPurgatory;
 import io.streamnative.pulsar.handlers.kop.utils.timer.SystemTimer;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.Node;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.errors.TopicExistsException;
@@ -52,6 +56,9 @@ class AdminManager {
 
     private final PulsarAdmin admin;
     private final int defaultNumPartitions;
+    private volatile Set<Node> brokersCache = new HashSet<>();
+    private final ReentrantReadWriteLock brokersCacheLock = new ReentrantReadWriteLock();
+
 
     public AdminManager(PulsarAdmin admin, KafkaServiceConfiguration conf) {
         this.admin = admin;
@@ -221,4 +228,17 @@ public Map<String, Errors> deleteTopics(Set<String> topicsToDelete) {
         });
         return result;
     }
+
+    public Collection<? extends Node> getBrokers() {
+        return brokersCache;
+    }
+
+    public void setBrokers(Set<Node> newBrokers) {
+        brokersCacheLock.writeLock().lock();
+        try {
+            this.brokersCache = newBrokers;
+        } finally {
+            brokersCacheLock.writeLock().unlock();
+        }
+    }
 }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/ChildChangeHandler.java

@@ -0,0 +1,57 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.kop;
+
+public interface ChildChangeHandler {
+    String path();
+
+    void handleChildChange();
+}
+
+class DeletionTopicsHandler implements ChildChangeHandler {
+    private final KopEventManager kopEventManager;
+
+    public DeletionTopicsHandler(KopEventManager kopEventManager) {
+        this.kopEventManager = kopEventManager;
+    }
+
+    @Override
+    public String path() {
+        return KopEventManager.getBrokersChangePath();
+    }
+
+    @Override
+    public void handleChildChange() {
+        kopEventManager.put(kopEventManager.getDeleteTopicEvent());
+    }
+}
+
+class BrokersChangeHandler implements ChildChangeHandler {
+    private final KopEventManager kopEventManager;
+
+    public BrokersChangeHandler(KopEventManager kopEventManager) {
+        this.kopEventManager = kopEventManager;
+    }
+
+    @Override
+    public String path() {
+        return KopEventManager.getBrokersChangePath();
+    }
+
+    @Override
+    public void handleChildChange() {
+        kopEventManager.put(kopEventManager.getBrokersChangeEvent());
+    }
+
+}

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java

@@ -95,6 +95,8 @@ public class KafkaProtocolHandler implements ProtocolHandler {
     private GroupCoordinator groupCoordinator;
     @Getter
     private TransactionCoordinator transactionCoordinator;
+    @Getter
+    private KopEventManager kopEventManager;
 
     /**
      * Listener for the changing of topic that stores offsets of consumer group.
@@ -285,6 +287,12 @@ public void start(BrokerService service) {
         ZooKeeperUtils.tryCreatePath(brokerService.pulsar().getZkClient(),
                 kafkaConfig.getGroupIdZooKeeperPath(), new byte[0]);
 
+        ZooKeeperUtils.tryCreatePath(brokerService.pulsar().getZkClient(),
+                KopEventManager.getKopPath(), new byte[0]);
+
+        ZooKeeperUtils.tryCreatePath(brokerService.pulsar().getZkClient(),
+                KopEventManager.getDeleteTopicsPath(), new byte[0]);
+
         PulsarAdmin pulsarAdmin;
         try {
             pulsarAdmin = brokerService.getPulsar().getAdminClient();
@@ -322,6 +330,12 @@ public void start(BrokerService service) {
 
         // init and start group coordinator
         startGroupCoordinator(pulsarClient);
+        // init KopEventManager
+        kopEventManager = new KopEventManager(groupCoordinator,
+                adminManager,
+                brokerService.getPulsar().getLocalMetadataStore());
+        kopEventManager.start();
+
         // and listener for Offset topics load/unload
         brokerService.pulsar()
                 .getNamespaceService()
@@ -425,13 +439,13 @@ public void startGroupCoordinator(PulsarClient pulsarClient) {
             .build();
 
         this.groupCoordinator = GroupCoordinator.of(
-            (PulsarClientImpl) pulsarClient,
-            groupConfig,
-            offsetConfig,
-            SystemTimer.builder()
-                .executorName("group-coordinator-timer")
-                .build(),
-            Time.SYSTEM
+                (PulsarClientImpl) pulsarClient,
+                groupConfig,
+                offsetConfig,
+                SystemTimer.builder()
+                        .executorName("group-coordinator-timer")
+                        .build(),
+                Time.SYSTEM
         );
         // always enable metadata expiration
         this.groupCoordinator.startup(true);
@@ -498,4 +512,5 @@ private void loadTxnLogTopics(TransactionCoordinator txnCoordinator) throws Exce
     public static @NonNull LookupClient getLookupClient(final PulsarService pulsarService) {
         return LOOKUP_CLIENT_MAP.computeIfAbsent(pulsarService, ignored -> new LookupClient(pulsarService));
     }
+
 }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

@@ -492,6 +492,8 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar,
         // Command response for all topics
         List<TopicMetadata> allTopicMetadata = Collections.synchronizedList(Lists.newArrayList());
         List<Node> allNodes = Collections.synchronizedList(Lists.newArrayList());
+        // Get all kop brokers in local cache
+        allNodes.addAll(adminManager.getBrokers());
 
         List<String> topics = metadataRequest.topics();
         // topics in format : persistent://%s/%s/abc-partition-x, will be grouped by as:
@@ -672,13 +674,12 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar,
             if (e != null) {
                 log.warn("[{}] Request {}: Exception fetching metadata, will return null Response",
                     ctx.channel(), metadataHar.getHeader(), e);
-                allNodes.add(newSelfNode());
                 MetadataResponse finalResponse =
-                    new MetadataResponse(
-                        allNodes,
-                        clusterName,
-                        controllerId,
-                        Collections.emptyList());
+                        new MetadataResponse(
+                                allNodes,
+                                clusterName,
+                                controllerId,
+                                Collections.emptyList());
                 resultFuture.complete(finalResponse);
                 return;
             }
@@ -687,13 +688,12 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar,
 
             if (topicsNumber == 0) {
                 // no topic partitions added, return now.
-                allNodes.add(newSelfNode());
                 MetadataResponse finalResponse =
-                    new MetadataResponse(
-                        allNodes,
-                        clusterName,
-                        controllerId,
-                        allTopicMetadata);
+                        new MetadataResponse(
+                                allNodes,
+                                clusterName,
+                                controllerId,
+                                allTopicMetadata);
                 resultFuture.complete(finalResponse);
                 return;
             }
@@ -2049,7 +2049,18 @@ protected void handleDeleteTopics(KafkaHeaderAndRequest deleteTopics,
         checkArgument(deleteTopics.getRequest() instanceof DeleteTopicsRequest);
         DeleteTopicsRequest request = (DeleteTopicsRequest) deleteTopics.getRequest();
         Set<String> topicsToDelete = request.topics();
-        resultFuture.complete(new DeleteTopicsResponse(adminManager.deleteTopics(topicsToDelete)));
+        Map<String, Errors> deleteTopicsResponse = adminManager.deleteTopics(topicsToDelete);
+
+        // create topic znode to trigger the coordinator DeleteTopicsEvent event
+        deleteTopicsResponse.forEach((topic, errors) -> {
+            if (errors == Errors.NONE) {
+                ZooKeeperUtils.tryCreatePath(pulsarService.getZkClient(),
+                        KopEventManager.getDeleteTopicsPath() + "/" + topic,
+                        new byte[0]);
+            }
+        });
+
+        resultFuture.complete(new DeleteTopicsResponse(deleteTopicsResponse));
     }
 
     /**