Bump pulsar version to 2.8.1.0

pom.xml

@@ -43,7 +43,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.15</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>
     <testng.version>6.14.3</testng.version>

tests/src/test/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/ProducerIdManagerTest.java

@@ -17,6 +17,8 @@
 import static org.testng.AssertJUnit.fail;
 
 import io.streamnative.pulsar.handlers.kop.KopProtocolHandlerTestBase;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.zookeeper.data.Stat;
 import org.testng.Assert;
@@ -31,6 +33,8 @@
 @Slf4j
 public class ProducerIdManagerTest extends KopProtocolHandlerTestBase {
 
+    private static final long DEFAULT_TEST_TIMEOUT = 20 * 1000;
+
     @BeforeClass
     @Override
     protected void setup() throws Exception {
@@ -51,7 +55,7 @@ protected void cleanZNode() throws Exception {
         }
     }
 
-    @Test
+    @Test(timeOut = DEFAULT_TEST_TIMEOUT)
     public void testGetProducerId() throws Exception {
         ProducerIdManager manager1 = new ProducerIdManager(0, pulsar.getLocalMetadataStore());
         manager1.initialize().get();
@@ -76,13 +80,14 @@ public void testGetProducerId() throws Exception {
         assertEquals(pid2 + ProducerIdManager.PID_BLOCK_SIZE * 2, manager2.generateProducerId().get().longValue());
     }
 
-    @Test
+    @Test(timeOut = DEFAULT_TEST_TIMEOUT)
     public void testExceedProducerIdLimit() throws Exception {
-        mockZooKeeper.create(ProducerIdManager.KOP_PID_BLOCK_ZNODE, null, null, null);
-        mockZooKeeper.setData(ProducerIdManager.KOP_PID_BLOCK_ZNODE,
-                ProducerIdManager.generateProducerIdBlockJson(
-                        new ProducerIdManager.ProducerIdBlock(
-                                1, Long.MAX_VALUE - ProducerIdManager.PID_BLOCK_SIZE, Long.MAX_VALUE)), -1);
+        pulsar.getLocalMetadataStore()
+                .put(ProducerIdManager.KOP_PID_BLOCK_ZNODE,
+                        ProducerIdManager.generateProducerIdBlockJson(
+                                new ProducerIdManager.ProducerIdBlock(
+                                        1, Long.MAX_VALUE - ProducerIdManager.PID_BLOCK_SIZE, Long.MAX_VALUE)),
+                        Optional.empty()).get(10, TimeUnit.SECONDS);
 
         ProducerIdManager producerIdManager = new ProducerIdManager(0, pulsar.getLocalMetadataStore());
         try {