[test] Fix unstable put kv test

Author: swuferhong

fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherITCase.java

@@ -21,6 +21,7 @@
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TableDescriptor;
 import com.alibaba.fluss.metadata.TableInfo;
+import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.record.KvRecordBatch;
 import com.alibaba.fluss.record.LogRecords;
 import com.alibaba.fluss.rpc.entity.FetchLogResultForBucket;
@@ -59,7 +60,11 @@
 import static com.alibaba.fluss.record.TestData.DATA_1_WITH_KEY_AND_VALUE;
 import static com.alibaba.fluss.record.TestData.EXPECTED_LOG_RESULTS_FOR_DATA_1_WITH_PK;
 import static com.alibaba.fluss.server.testutils.KvTestUtils.assertLookupResponse;
+import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.assertFetchLogResponseWithRowKind;
+import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.createTable;
+import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.newFetchLogRequest;
 import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.newLookupRequest;
+import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.newPutKvRequest;
 import static com.alibaba.fluss.testutils.DataTestUtils.assertLogRecordsEquals;
 import static com.alibaba.fluss.testutils.DataTestUtils.assertLogRecordsEqualsWithRowKind;
 import static com.alibaba.fluss.testutils.DataTestUtils.genKvRecordBatch;
@@ -103,7 +108,7 @@ void testProduceLogNeedAck() throws Exception {
         FLUSS_CLUSTER_EXTENSION.waitUtilAllGatewayHasSameMetadata();
 
         long tableId =
-                RpcMessageTestUtils.createTable(
+                createTable(
                         FLUSS_CLUSTER_EXTENSION,
                         DATA1_TABLE_PATH,
                         data1NonPkTableInfo.getTableDescriptor());
@@ -131,9 +136,7 @@ void testProduceLogNeedAck() throws Exception {
 
         // check leader log data.
         RpcMessageTestUtils.assertFetchLogResponse(
-                leaderGateWay
-                        .fetchLog(RpcMessageTestUtils.newFetchLogRequest(-1, tableId, bucketId, 0L))
-                        .get(),
+                leaderGateWay.fetchLog(newFetchLogRequest(-1, tableId, bucketId, 0L)).get(),
                 tableId,
                 bucketId,
                 10L,
@@ -178,17 +181,23 @@ void testProduceLogNeedAck() throws Exception {
     @Test
     void testPutKvNeedAck() throws Exception {
         // set bucket count to 1 to easy for debug.
-        TableInfo data1PkTableInfo = createPkTable();
+        TablePath tp = TablePath.of("test_db_1", "test_pk_table_need_ack");
+        TableInfo data1PkTableInfo =
+                new TableInfo(
+                        tp,
+                        DATA1_TABLE_ID_PK,
+                        TableDescriptor.builder()
+                                .schema(DATA1_SCHEMA_PK)
+                                .distributedBy(1, "a")
+                                .build(),
+                        1);
 
         // wait until all the gateway has same metadata because the follower fetcher manager need
         // to get the leader address from server metadata while make follower.
         FLUSS_CLUSTER_EXTENSION.waitUtilAllGatewayHasSameMetadata();
 
         long tableId =
-                RpcMessageTestUtils.createTable(
-                        FLUSS_CLUSTER_EXTENSION,
-                        DATA1_TABLE_PATH_PK,
-                        data1PkTableInfo.getTableDescriptor());
+                createTable(FLUSS_CLUSTER_EXTENSION, tp, data1PkTableInfo.getTableDescriptor());
         int bucketId = 0;
         TableBucket tb = new TableBucket(tableId, bucketId);
 
@@ -202,7 +211,7 @@ void testPutKvNeedAck() throws Exception {
         assertPutKvResponse(
                 leaderGateWay
                         .putKv(
-                                RpcMessageTestUtils.newPutKvRequest(
+                                newPutKvRequest(
                                         tableId,
                                         bucketId,
                                         -1,
@@ -211,10 +220,8 @@ void testPutKvNeedAck() throws Exception {
                 bucketId);
 
         // check leader log data.
-        RpcMessageTestUtils.assertFetchLogResponseWithRowKind(
-                leaderGateWay
-                        .fetchLog(RpcMessageTestUtils.newFetchLogRequest(-1, tableId, bucketId, 0L))
-                        .get(),
+        assertFetchLogResponseWithRowKind(
+                leaderGateWay.fetchLog(newFetchLogRequest(-1, tableId, bucketId, 0L)).get(),
                 tableId,
                 bucketId,
                 8L,
@@ -263,7 +270,7 @@ void testFlushForPutKvNeedAck() throws Exception {
 
         // create a table and wait all replica ready
         long tableId =
-                RpcMessageTestUtils.createTable(
+                createTable(
                         FLUSS_CLUSTER_EXTENSION,
                         DATA1_TABLE_PATH_PK,
                         data1PkTableInfo.getTableDescriptor());
@@ -298,8 +305,7 @@ void testFlushForPutKvNeedAck() throws Exception {
                         Tuple2.of("k2", new Object[] {3, "b1"}));
 
         CompletableFuture<PutKvResponse> putResponse =
-                leaderGateWay.putKv(
-                        RpcMessageTestUtils.newPutKvRequest(tableId, bucketId, -1, kvRecords));
+                leaderGateWay.putKv(newPutKvRequest(tableId, bucketId, -1, kvRecords));
 
         // wait util the log has been written
         Replica replica = FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tb);