Issue #61 : Fix unload and close broker error

src/main/java/io/streamnative/kop/InternalProducer.java

@@ -0,0 +1,53 @@
+/**
+ * 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.kop;
+
+import java.util.concurrent.CompletableFuture;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.service.Producer;
+import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.service.Topic;
+
+/**
+ * InternalServerCnx, this only used to construct internalProducer / internalConsumer.
+ * So when topic is unload, we could disconnect the connection between kafkaRequestHandler and client,
+ * by internalProducer / internalConsumer.close();
+ * which means when topic unload happens, we should close the connection.
+ */
+@Slf4j
+public class InternalProducer extends Producer {
+    public InternalProducer(Topic topic, ServerCnx cnx,
+                            long producerId, String producerName) {
+        super(topic, cnx, producerId, producerName, null,
+            false, null, null);
+    }
+
+    // this will call back by bundle unload
+    @Override
+    public CompletableFuture<Void> disconnect() {
+        InternalServerCnx cnx = (InternalServerCnx) getCnx();
+        CompletableFuture<Void> future = new CompletableFuture<>();
+
+        cnx.getBrokerService().executor().execute(() -> {
+            log.info("Disconnecting producer: {}", this);
+            getTopic().removeProducer(this);
+            cnx.closeProducer(this);
+            future.complete(null);
+        });
+
+        return future;
+    }
+
+
+}

src/main/java/io/streamnative/kop/InternalServerCnx.java

@@ -0,0 +1,63 @@
+/**
+ * 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.kop;
+
+import java.net.InetSocketAddress;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.service.Producer;
+import org.apache.pulsar.broker.service.ServerCnx;
+
+/**
+ * InternalServerCnx, this only used to construct internalProducer / internalConsumer.
+ * So when topic is unload, we could disconnect the connection between kafkaRequestHandler and client,
+ * by internalProducer / internalConsumer.close();
+ * which means when topic unload happens, we should close the connection.
+ */
+@Slf4j
+public class InternalServerCnx extends ServerCnx {
+    @Getter
+    KafkaRequestHandler kafkaRequestHandler;
+
+    public InternalServerCnx(KafkaRequestHandler kafkaRequestHandler) {
+        super(kafkaRequestHandler.getPulsarService());
+        this.kafkaRequestHandler = kafkaRequestHandler;
+        // this is the client address that connect to this server.
+        this.remoteAddress = kafkaRequestHandler.remoteAddress;
+
+        // mock some values, or Producer create will meet NPE.
+        // used in test, which will not call channel.active, and not call updateCtx.
+        if (this.remoteAddress == null) {
+            this.remoteAddress = new InetSocketAddress("localhost", 9999);
+        }
+    }
+
+    // this will call back by bundle unload
+    @Override
+    public void closeProducer(Producer producer) {
+        // removes producer-connection from map and send close command to producer
+        if (log.isDebugEnabled()) {
+            log.debug("[{}] Removed topic: {}'s producer: {}.",
+                remoteAddress, producer.getTopic().getName(), producer);
+        }
+
+        kafkaRequestHandler.close();
+    }
+
+    // called after channel active
+    public void updateCtx() {
+        this.remoteAddress = kafkaRequestHandler.remoteAddress;
+    }
+
+}

src/main/java/io/streamnative/kop/KafkaChannelInitializer.java

@@ -38,8 +38,6 @@ public class KafkaChannelInitializer extends ChannelInitializer<SocketChannel> {
     @Getter
     private final KafkaServiceConfiguration kafkaConfig;
     @Getter
-    private final KafkaTopicManager kafkaTopicManager;
-    @Getter
     private final GroupCoordinator groupCoordinator;
     @Getter
     private final boolean enableTls;
@@ -48,13 +46,11 @@ public class KafkaChannelInitializer extends ChannelInitializer<SocketChannel> {
 
     public KafkaChannelInitializer(PulsarService pulsarService,
                                    KafkaServiceConfiguration kafkaConfig,
-                                   KafkaTopicManager kafkaTopicManager,
                                    GroupCoordinator groupCoordinator,
                                    boolean enableTLS) throws Exception {
         super();
         this.pulsarService = pulsarService;
         this.kafkaConfig = kafkaConfig;
-        this.kafkaTopicManager = kafkaTopicManager;
         this.groupCoordinator = groupCoordinator;
         this.enableTls = enableTLS;
 
@@ -74,7 +70,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast("frameDecoder",
             new LengthFieldBasedFrameDecoder(MAX_FRAME_LENGTH, 0, 4, 0, 4));
         ch.pipeline().addLast("handler",
-            new KafkaRequestHandler(pulsarService, kafkaConfig, kafkaTopicManager, groupCoordinator, enableTls));
+            new KafkaRequestHandler(pulsarService, kafkaConfig, groupCoordinator, enableTls));
     }
 
 }