[server] Fix admin operation cannot be done error when bootstrap server config as one of TabletServer (apache#525)

Author: swuferhong

fluss-client/src/test/java/com/alibaba/fluss/client/admin/FlussAdminITCase.java

@@ -16,13 +16,16 @@
 
 package com.alibaba.fluss.client.admin;
 
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
 import com.alibaba.fluss.client.metadata.KvSnapshotMetadata;
 import com.alibaba.fluss.client.metadata.KvSnapshots;
 import com.alibaba.fluss.client.table.Table;
 import com.alibaba.fluss.client.table.writer.UpsertWriter;
 import com.alibaba.fluss.cluster.ServerNode;
 import com.alibaba.fluss.config.AutoPartitionTimeUnit;
 import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.exception.DatabaseAlreadyExistException;
 import com.alibaba.fluss.exception.DatabaseNotEmptyException;
 import com.alibaba.fluss.exception.DatabaseNotExistException;
@@ -742,6 +745,36 @@ tablePath, newPartitionSpec("pt", String.valueOf(currentYear + 1)), false)
                         String.valueOf(currentYear)));
     }
 
+    @Test
+    void testBootstrapServerConfigAsTabletServer() throws Exception {
+        Configuration newConf = clientConf;
+        ServerNode ts0 = FLUSS_CLUSTER_EXTENSION.getTabletServerNodes().get(0);
+        newConf.set(
+                ConfigOptions.BOOTSTRAP_SERVERS,
+                Collections.singletonList(String.format("%s:%d", ts0.host(), ts0.port())));
+        try (Connection conn = ConnectionFactory.createConnection(clientConf)) {
+            Admin newAdmin = conn.getAdmin();
+            String dbName = "test_bootstrap_server_t1";
+            newAdmin.createDatabase(
+                            dbName,
+                            DatabaseDescriptor.builder().comment("test comment").build(),
+                            false)
+                    .get();
+            newAdmin.createTable(
+                            TablePath.of(dbName, "test_table_1"),
+                            TableDescriptor.builder().schema(Schema.newBuilder().build()).build(),
+                            false)
+                    .get();
+            assertThat(newAdmin.getDatabaseInfo(dbName).get().getDatabaseName()).isEqualTo(dbName);
+            assertThat(
+                            newAdmin.getTableInfo(TablePath.of(dbName, "test_table_1"))
+                                    .get()
+                                    .getTablePath()
+                                    .getTableName())
+                    .isEqualTo("test_table_1");
+        }
+    }
+
     private void assertHasTabletServerNumber(int tabletServerNumber) {
         CoordinatorGateway coordinatorGateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
         retry(

fluss-client/src/test/java/com/alibaba/fluss/client/write/RecordAccumulatorTest.java

@@ -557,7 +557,7 @@ private Cluster updateCluster(List<BucketLocation> bucketLocations) {
 
         return new Cluster(
                 aliveTabletServersById,
-                new ServerNode(-1, "localhost", 89, ServerType.COORDINATOR),
+                new ServerNode(0, "localhost", 89, ServerType.COORDINATOR),
                 bucketsByPath,
                 tableIdByPath,
                 Collections.emptyMap(),

fluss-client/src/test/java/com/alibaba/fluss/client/write/StickyStaticBucketAssignerTest.java

@@ -224,7 +224,7 @@ private Cluster updateCluster(List<BucketLocation> bucketLocations) {
 
         return new Cluster(
                 aliveTabletServersById,
-                new ServerNode(-1, "localhost", 89, ServerType.COORDINATOR),
+                new ServerNode(0, "localhost", 89, ServerType.COORDINATOR),
                 bucketsByPath,
                 tableIdByPath,
                 Collections.emptyMap(),

fluss-common/src/test/java/com/alibaba/fluss/cluster/ClusterTest.java

@@ -46,7 +46,7 @@
 /** Test for {@link Cluster}. */
 class ClusterTest {
     private static final ServerNode COORDINATOR_SERVER =
-            new ServerNode(-1, "localhost", 98, ServerType.COORDINATOR);
+            new ServerNode(0, "localhost", 98, ServerType.COORDINATOR);
     private static final ServerNode[] NODES =
             new ServerNode[] {
                 new ServerNode(0, "localhost", 99, ServerType.TABLET_SERVER),

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -274,10 +274,11 @@ private ServerNode getCoordinatorServerNode() {
                     .getCoordinatorAddress()
                     .map(
                             coordinatorAddress ->
-                                    // we set id to -1 as the id for the id stored in zk
-                                    // for coordinator server is an uuid now
+                                    // TODO we set id to 0 as that CoordinatorServer don't support
+                                    // HA, if we support HA, we need to set id to the config
+                                    // CoordinatorServer id to avoid node drift.
                                     new ServerNode(
-                                            -1,
+                                            0,
                                             coordinatorAddress.getHost(),
                                             coordinatorAddress.getPort(),
                                             ServerType.COORDINATOR))

fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTestBase.java

@@ -201,7 +201,7 @@ public void setup() throws Exception {
     private void initMetadataCache(ServerMetadataCache metadataCache) {
         metadataCache.updateMetadata(
                 new ClusterMetadataInfo(
-                        Optional.of(new ServerNode(-1, "localhost", 1234, ServerType.COORDINATOR)),
+                        Optional.of(new ServerNode(0, "localhost", 1234, ServerType.COORDINATOR)),
                         new HashSet<>(
                                 Arrays.asList(
                                         new ServerNode(

fluss-server/src/test/java/com/alibaba/fluss/server/testutils/FlussClusterExtension.java

@@ -222,9 +222,9 @@ public void startCoordinatorServer() throws Exception {
                 coordinatorServer = new CoordinatorServer(conf);
                 coordinatorServer.start();
                 coordinatorServerNode =
-                        // we use -1 as coordinator server id
+                        // TODO, Currently, we use 0 as coordinator server id.
                         new ServerNode(
-                                -1, HOST_ADDRESS, availablePort.getPort(), ServerType.COORDINATOR);
+                                0, HOST_ADDRESS, availablePort.getPort(), ServerType.COORDINATOR);
             }
         } else {
             // start the existing coordinator server