diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/InternalServerCnx.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/InternalServerCnx.java
index 195f256d84..8142c21809 100644
--- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/InternalServerCnx.java
+++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/InternalServerCnx.java
@@ -62,6 +62,7 @@ public void closeProducer(Producer producer) {
     // called after channel active
     public void updateCtx(final SocketAddress remoteAddress) {
         this.remoteAddress = remoteAddress;
+        this.ctx = kafkaRequestHandler.ctx;
     }
 
     @Override
@@ -78,4 +79,14 @@ public void disableCnxAutoRead() {
     public void cancelPublishBufferLimiting() {
         // do nothing is this mock
     }
+
+    @Override
+    public boolean equals(Object o) {
+        return super.equals(o);
+    }
+
+    @Override
+    public int hashCode() {
+        return super.hashCode();
+    }
 }
diff --git a/pom.xml b/pom.xml
index e66186c5c6..b0a68954f1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -46,7 +46,7 @@
     <lombok.version>1.18.4</lombok.version>
     <mockito.version>2.22.0</mockito.version>
     <pulsar.group.id>io.streamnative</pulsar.group.id>
-    <pulsar.version>2.8.0.3</pulsar.version>
+    <pulsar.version>2.8.1.0</pulsar.version>
     <slf4j.version>1.7.25</slf4j.version>
     <spotbugs-annotations.version>3.1.8</spotbugs-annotations.version>
     <testcontainers.version>1.15.1</testcontainers.version>