fix(interactive): increase write Kafka speed by using async method (#4549)

Author: siyuan0322

interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/coordinator/GarbageCollectManager.java

@@ -47,7 +47,7 @@ public void start() {
                             for (int i = 0; i < CommonConfig.STORE_NODE_COUNT.get(configs); i++) {
                                 CoordinatorSnapshotClient client = clients.getClient(i);
                                 client.synchronizeSnapshot(offlineVersion);
-                                if (i == 0 && offlineVersion % 1000 == 0) {
+                                if (i == 0) {
                                     logger.info("Offline version updated to {}", offlineVersion);
                                 }
                             }
@@ -63,6 +63,7 @@ public void start() {
                 interval,
                 interval,
                 TimeUnit.MILLISECONDS);
+        logger.info("GarbageCollectManaget started");
     }
 
     public void stop() {

interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/frontend/write/KafkaAppender.java

@@ -161,7 +161,7 @@ private long processTask(LogWriter logWriter, IngestTask task) throws IOExceptio
                     OperationBatch batch = entry.getValue().build();
                     // logger.info("Log writer append storeId [{}], batch size: {}", storeId,
                     // batch.getOperationCount());
-                    logWriter.append(storeId, new LogEntry(ingestSnapshotId.get(), batch));
+                    logWriter.appendAsync(storeId, new LogEntry(ingestSnapshotId.get(), batch));
                 }
             } catch (Exception e) {
                 // write failed, just throw out to fail this task
@@ -256,7 +256,7 @@ public List<Long> replayDMLRecordsFrom(long offset, long timestamp) throws IOExc
                         if (batch.getOperationCount() == 0) {
                             continue;
                         }
-                        logWriter.append(storeId, new LogEntry(batchSnapshotId, batch));
+                        logWriter.appendAsync(storeId, new LogEntry(batchSnapshotId, batch));
                         replayCount++;
                     }
                     ids.add(batchSnapshotId + 1);

interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/store/StoreSnapshotService.java

@@ -25,6 +25,7 @@ public void synchronizeMinQuerySnapshotId(
             SynchronizeMinQuerySnapshotIdRequest request,
             StreamObserver<SynchronizeMinQuerySnapshotIdResponse> responseObserver) {
         long snapshotId = request.getSnapshotId();
+        logger.info("synchronize min query snapshot id [{}]", snapshotId);
         this.storeService.garbageCollect(
                 snapshotId,
                 new CompletionCallback<Void>() {

interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/LogWriter.java

@@ -34,6 +34,8 @@ public interface LogWriter extends AutoCloseable {
 
     long append(int partition, LogEntry logEntry) throws IOException;
 
+    public Future<RecordMetadata> appendAsync(LogEntry logEntry) throws IOException;
+
     public Future<RecordMetadata> appendAsync(int partition, LogEntry logEntry) throws IOException;
 
     void close() throws IOException;

interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogWriter.java

@@ -67,11 +67,18 @@ public long append(LogEntry logEntry) {
     }
 
     @Override
-    public Future<RecordMetadata> appendAsync(int partition, LogEntry logEntry) {
+    public Future<RecordMetadata> appendAsync(LogEntry logEntry) {
         Future<RecordMetadata> future = producer.send(new ProducerRecord<>(topicName, logEntry));
         return future;
     }
 
+    @Override
+    public Future<RecordMetadata> appendAsync(int partition, LogEntry logEntry) {
+        Future<RecordMetadata> future =
+                producer.send(new ProducerRecord<>(topicName, partition, null, logEntry));
+        return future;
+    }
+
     public long waitFuture(Future<RecordMetadata> future) {
         RecordMetadata recordMetadata;
         try {

interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/readonly/ReadOnlyLogWriter.java

@@ -36,6 +36,10 @@ public long append(int partition, LogEntry logEntry) throws IOException {
         return offset;
     }
 
+    public Future<RecordMetadata> appendAsync(LogEntry logEntry) throws IOException {
+        return null;
+    }
+
     public Future<RecordMetadata> appendAsync(int partition, LogEntry logEntry) throws IOException {
         return null;
     }