Use MetadataStore instead of mockZooKeeper in ProducerIdManagerTest (#…

…951)

Fix https://github.com/streamnative/kop/runs/4425667410?check_suite_focus=true#step:5:283

#950

### Motivation

MockZookeeper API was changed in apache/pulsar#13066, the `CreateMode` shouldn't be null. We should use MetadataStore instead of mockZooKeeper since KoP already remove all zkClient usage.

### Modifications

Use MetadataStore instead of mockZooKeeper in ProducerIdManagerTest

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

@@ -17,8 +17,9 @@
 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;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -31,6 +32,8 @@
 @Slf4j
 public class ProducerIdManagerTest extends KopProtocolHandlerTestBase {
 
+    private static final long DEFAULT_TEST_TIMEOUT = 20 * 1000;
+
     @BeforeClass
     @Override
     protected void setup() throws Exception {
@@ -43,15 +46,13 @@ protected void cleanup() throws Exception {
         super.internalCleanup();
     }
 
-    @BeforeMethod
+    @BeforeMethod(timeOut = DEFAULT_TEST_TIMEOUT)
     protected void cleanZNode() throws Exception {
-        Stat stat = mockZooKeeper.exists(ProducerIdManager.KOP_PID_BLOCK_ZNODE, null);
-        if (stat != null) {
-            mockZooKeeper.delete(ProducerIdManager.KOP_PID_BLOCK_ZNODE, -1);
-        }
+        pulsar.getLocalMetadataStore()
+                .deleteRecursive(ProducerIdManager.KOP_PID_BLOCK_ZNODE).get(10, TimeUnit.SECONDS);
     }
 
-    @Test
+    @Test(timeOut = DEFAULT_TEST_TIMEOUT)
     public void testGetProducerId() throws Exception {
         ProducerIdManager manager1 = new ProducerIdManager(0, pulsar.getLocalMetadataStore());
         manager1.initialize().get();
@@ -76,13 +77,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 {