fix error

Author: zcoo

fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java

@@ -214,7 +214,7 @@ public void registerCoordinatorServer(CoordinatorAddress coordinatorAddress) thr
      * registerCoordinatorLeader(). This is to ensure the coordinator get and update the coordinator
      * epoch and coordinator epoch zk version.
      */
-    public Optional<ZkEpoch> fenceBecomeCoordinatorLeader(int coordinatorId) throws Exception {
+    public Optional<ZkEpoch> fenceBecomeCoordinatorLeader(String coordinatorId) throws Exception {
         ensureEpochZnodeExists();
 
         try {

fluss-server/src/main/java/org/apache/fluss/server/zk/data/CoordinatorAddressJsonSerde.java

@@ -35,7 +35,7 @@ public class CoordinatorAddressJsonSerde
 
     public static final CoordinatorAddressJsonSerde INSTANCE = new CoordinatorAddressJsonSerde();
     private static final String VERSION_KEY = "version";
-    private static final int VERSION = 3;
+    private static final int VERSION = 2;
 
     private static final String ID = "id";
     private static final String HOST = "host";
@@ -51,7 +51,7 @@ public void serialize(CoordinatorAddress coordinatorAddress, JsonGenerator gener
             throws IOException {
         generator.writeStartObject();
         writeVersion(generator);
-        generator.writeNumberField(ID, coordinatorAddress.getId());
+        generator.writeStringField(ID, coordinatorAddress.getId());
         generator.writeStringField(
                 LISTENERS, Endpoint.toListenersString(coordinatorAddress.getEndpoints()));
         generator.writeEndObject();
@@ -60,18 +60,13 @@ public void serialize(CoordinatorAddress coordinatorAddress, JsonGenerator gener
     @Override
     public CoordinatorAddress deserialize(JsonNode node) {
         int version = node.get(VERSION_KEY).asInt();
-        int id;
+        String id = node.get(ID).asText();
         List<Endpoint> endpoints;
         if (version == 1) {
-            id = Integer.parseInt(node.get(ID).asText());
             String host = node.get(HOST).asText();
             int port = node.get(PORT).asInt();
             endpoints = Collections.singletonList(new Endpoint(host, port, "CLIENT"));
-        } else if (version == 2) {
-            id = Integer.parseInt(node.get(ID).asText());
-            endpoints = Endpoint.fromListenersString(node.get(LISTENERS).asText());
         } else {
-            id = node.get(ID).asInt();
             endpoints = Endpoint.fromListenersString(node.get(LISTENERS).asText());
         }
         return new CoordinatorAddress(id, endpoints);

fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java

@@ -174,7 +174,7 @@ static void baseBeforeAll() throws Exception {
         // register coordinator server
         zookeeperClient.registerCoordinatorLeader(
                 new CoordinatorAddress(
-                        2, Endpoint.fromListenersString("CLIENT://localhost:10012")));
+                        "2", Endpoint.fromListenersString("CLIENT://localhost:10012")));
 
         // register 3 tablet servers
         for (int i = 0; i < 3; i++) {

fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java

@@ -113,7 +113,7 @@ void testCoordinatorLeader() throws Exception {
         assertThat(zookeeperClient.getCoordinatorLeaderAddress()).isEmpty();
         CoordinatorAddress coordinatorAddress =
                 new CoordinatorAddress(
-                        2, Endpoint.fromListenersString("CLIENT://localhost1:10012"));
+                        "2", Endpoint.fromListenersString("CLIENT://localhost1:10012"));
         // register leader address
         zookeeperClient.registerCoordinatorLeader(coordinatorAddress);
         // check get leader address

fluss-server/src/test/java/org/apache/fluss/server/zk/data/CoordinatorAddressJsonSerdeTest.java

@@ -39,7 +39,7 @@ public class CoordinatorAddressJsonSerdeTest extends JsonSerdeTestBase<Coordinat
     protected CoordinatorAddress[] createObjects() {
         CoordinatorAddress coordinatorAddress =
                 new CoordinatorAddress(
-                        1,
+                        "1",
                         Arrays.asList(
                                 new Endpoint("localhost", 1001, "CLIENT"),
                                 new Endpoint("127.0.0.1", 9124, "FLUSS")));
@@ -49,38 +49,23 @@ protected CoordinatorAddress[] createObjects() {
     @Override
     protected String[] expectedJsons() {
         return new String[] {
-            "{\"version\":3,\"id\":1,\"listeners\":\"CLIENT://localhost:1001,FLUSS://127.0.0.1:9124\"}"
+            "{\"version\":2,\"id\":\"1\",\"listeners\":\"CLIENT://localhost:1001,FLUSS://127.0.0.1:9124\"}"
         };
     }
 
     @Test
     void testCompatibility() throws IOException {
-        // version 1
         JsonNode jsonInVersion1 =
                 new ObjectMapper()
                         .readTree(
                                 "{\"version\":1,\"id\":\"1\",\"host\":\"localhost\",\"port\":1001}"
                                         .getBytes(StandardCharsets.UTF_8));
 
-        CoordinatorAddress coordinatorAddress1 =
+        CoordinatorAddress coordinatorAddress =
                 CoordinatorAddressJsonSerde.INSTANCE.deserialize(jsonInVersion1);
-        CoordinatorAddress expectedCoordinator1 =
-                new CoordinatorAddress(1, Endpoint.fromListenersString("CLIENT://localhost:1001"));
-        assertEquals(coordinatorAddress1, expectedCoordinator1);
-
-        // version 2
-        JsonNode jsonInVersion2 =
-                new ObjectMapper()
-                        .readTree(
-                                "{\"version\":2,\"id\":\"2\",\"listeners\":\"CLIENT://localhost:1001,FLUSS://127.0.0.1:9124\"}"
-                                        .getBytes(StandardCharsets.UTF_8));
-        CoordinatorAddress coordinatorAddress2 =
-                CoordinatorAddressJsonSerde.INSTANCE.deserialize(jsonInVersion2);
-        CoordinatorAddress expectedCoordinator2 =
+        CoordinatorAddress expectedCoordinator =
                 new CoordinatorAddress(
-                        2,
-                        Endpoint.fromListenersString(
-                                "CLIENT://localhost:1001,FLUSS://127.0.0.1:9124"));
-        assertEquals(coordinatorAddress2, expectedCoordinator2);
+                        "1", Endpoint.fromListenersString("CLIENT://localhost:1001"));
+        assertEquals(coordinatorAddress, expectedCoordinator);
     }
 }