fix(interactive): The replay doesn't need to fill the kafka again. (#4296)

Author: siyuan0322

interactive_engine/common/src/main/java/com/alibaba/graphscope/groot/common/config/CoordinatorConfig.java

@@ -17,7 +17,7 @@
 
 public class CoordinatorConfig {
     public static final Config<Long> SNAPSHOT_INCREASE_INTERVAL_MS =
-            Config.longConfig("snapshot.increase.interval.ms", 1000L);
+            Config.longConfig("snapshot.increase.interval.ms", 2000L);
 
     public static final Config<Long> OFFSETS_PERSIST_INTERVAL_MS =
             Config.longConfig("offsets.persist.interval.ms", 1000L);

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/client/ExecutionClient.java

@@ -21,6 +21,7 @@
 import com.alibaba.graphscope.common.client.type.ExecutionResponseListener;
 import com.alibaba.graphscope.common.config.Configs;
 import com.alibaba.graphscope.common.config.QueryTimeoutConfig;
+import com.alibaba.graphscope.gremlin.plugin.QueryLogger;
 
 /**
  * client to submit request to remote engine service
@@ -37,7 +38,8 @@ public ExecutionClient(ChannelFetcher<C> channelFetcher) {
     public abstract void submit(
             ExecutionRequest request,
             ExecutionResponseListener listener,
-            QueryTimeoutConfig timeoutConfig)
+            QueryTimeoutConfig timeoutConfig,
+            QueryLogger queryLogger)
             throws Exception;
 
     public abstract void close() throws Exception;

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/client/HttpExecutionClient.java

@@ -25,6 +25,7 @@
 import com.alibaba.graphscope.gaia.proto.GraphAlgebraPhysical;
 import com.alibaba.graphscope.gaia.proto.IrResult;
 import com.alibaba.graphscope.gaia.proto.StoredProcedure;
+import com.alibaba.graphscope.gremlin.plugin.QueryLogger;
 import com.alibaba.graphscope.interactive.client.Session;
 import com.alibaba.graphscope.interactive.client.common.Result;
 import com.alibaba.graphscope.interactive.client.impl.DefaultSession;
@@ -56,7 +57,8 @@ public HttpExecutionClient(Configs graphConfig, ChannelFetcher<URI> channelFetch
     public void submit(
             ExecutionRequest request,
             ExecutionResponseListener listener,
-            QueryTimeoutConfig timeoutConfig)
+            QueryTimeoutConfig timeoutConfig,
+            QueryLogger queryLogger)
             throws Exception {
         List<CompletableFuture> responseFutures = Lists.newArrayList();
         for (URI httpURI : channelFetcher.fetch()) {

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/client/RpcExecutionClient.java

@@ -23,6 +23,7 @@
 import com.alibaba.graphscope.common.config.PegasusConfig;
 import com.alibaba.graphscope.common.config.QueryTimeoutConfig;
 import com.alibaba.graphscope.gaia.proto.IrResult;
+import com.alibaba.graphscope.gremlin.plugin.QueryLogger;
 import com.alibaba.pegasus.RpcChannel;
 import com.alibaba.pegasus.RpcClient;
 import com.alibaba.pegasus.intf.ResultProcessor;
@@ -54,7 +55,8 @@ public RpcExecutionClient(Configs graphConfig, ChannelFetcher<RpcChannel> channe
     public void submit(
             ExecutionRequest request,
             ExecutionResponseListener listener,
-            QueryTimeoutConfig timeoutConfig)
+            QueryTimeoutConfig timeoutConfig,
+            QueryLogger queryLogger)
             throws Exception {
         if (rpcClientRef.get() == null) {
             rpcClientRef.compareAndSet(null, new RpcClient(channelFetcher.fetch()));
@@ -97,12 +99,13 @@ public void process(PegasusClient.JobResponse jobResponse) {
                     @Override
                     public void finish() {
                         listener.onCompleted();
-                        logger.info("[compile]: received results from engine");
+                        queryLogger.info("[compile]: received results from engine");
                     }
 
                     @Override
                     public void error(Status status) {
                         listener.onError(status.asException());
+                        queryLogger.error("[compile]: fail to receive results from engine");
                     }
                 },
                 timeoutConfig.getChannelTimeoutMS());

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/fetcher/DynamicIrMetaFetcher.java

@@ -20,7 +20,7 @@
 
 import com.alibaba.graphscope.common.config.Configs;
 import com.alibaba.graphscope.common.config.GraphConfig;
-import com.alibaba.graphscope.common.ir.meta.GraphId;
+import com.alibaba.graphscope.common.config.PlannerConfig;
 import com.alibaba.graphscope.common.ir.meta.IrMeta;
 import com.alibaba.graphscope.common.ir.meta.IrMetaStats;
 import com.alibaba.graphscope.common.ir.meta.IrMetaTracker;
@@ -46,21 +46,27 @@ public class DynamicIrMetaFetcher extends IrMetaFetcher implements AutoCloseable
     private volatile IrMetaStats currentState;
     // To manage the state changes of statistics resulting from different update operations.
     private volatile StatsState statsState;
-    private volatile Boolean statsEnabled = null;
+    private final boolean fetchStats;
 
     public DynamicIrMetaFetcher(Configs configs, IrMetaReader dataReader, IrMetaTracker tracker) {
         super(dataReader, tracker);
-        this.scheduler = new ScheduledThreadPoolExecutor(2);
+        this.scheduler = new ScheduledThreadPoolExecutor(1);
         this.scheduler.scheduleAtFixedRate(
                 () -> syncMeta(),
-                2000,
+                0,
                 GraphConfig.GRAPH_META_SCHEMA_FETCH_INTERVAL_MS.get(configs),
                 TimeUnit.MILLISECONDS);
-        this.scheduler.scheduleAtFixedRate(
-                () -> syncStats(statsEnabled == null ? false : statsEnabled),
-                2000,
-                GraphConfig.GRAPH_META_STATISTICS_FETCH_INTERVAL_MS.get(configs),
-                TimeUnit.MILLISECONDS);
+        this.fetchStats =
+                PlannerConfig.GRAPH_PLANNER_IS_ON.get(configs)
+                        && PlannerConfig.GRAPH_PLANNER_OPT.get(configs).equals("CBO");
+        if (this.fetchStats) {
+            logger.info("start to schedule statistics fetch task");
+            this.scheduler.scheduleAtFixedRate(
+                    () -> syncStats(),
+                    2000,
+                    GraphConfig.GRAPH_META_STATISTICS_FETCH_INTERVAL_MS.get(configs),
+                    TimeUnit.MILLISECONDS);
+        }
     }
 
     @Override
@@ -94,32 +100,18 @@ private synchronized void syncMeta() {
                             meta.getSchema(),
                             meta.getStoredProcedures(),
                             curStats);
-            boolean statsEnabled = getStatsEnabled(this.currentState.getGraphId());
-            if (statsEnabled && this.statsState != StatsState.SYNCED
-                    || (!statsEnabled && this.statsState != StatsState.MOCKED)) {
-                logger.debug("start to sync stats");
-                syncStats(statsEnabled);
+            if (this.fetchStats && this.statsState != StatsState.SYNCED) {
+                logger.info("start to schedule statistics fetch task");
+                syncStats();
             }
         } catch (Throwable e) {
-            logger.warn("failed to read meta data, error is {}", e);
-        }
-    }
-
-    private boolean getStatsEnabled(GraphId graphId) {
-        try {
-            return this.statsEnabled == null
-                    ? this.reader.syncStatsEnabled(graphId)
-                    : this.statsEnabled;
-        } catch (
-                Throwable e) { // if errors happen when reading stats enabled, we assume it is false
-            logger.warn("failed to read stats enabled, error is {}", e);
-            return false;
+            logger.warn("failed to read meta data", e);
         }
     }
 
-    private synchronized void syncStats(boolean statsEnabled) {
+    private synchronized void syncStats() {
         try {
-            if (this.currentState != null && statsEnabled) {
+            if (this.currentState != null) {
                 GraphStatistics stats = this.reader.readStats(this.currentState.getGraphId());
                 logger.debug("statistics from remote: {}", stats);
                 if (stats != null && stats.getVertexCount() != 0) {
@@ -137,7 +129,7 @@ private synchronized void syncStats(boolean statsEnabled) {
                 }
             }
         } catch (Throwable e) {
-            logger.warn("failed to read graph statistics, error is {}", e);
+            logger.warn("failed to read graph statistics, error is", e);
         } finally {
             try {
                 if (this.currentState != null
@@ -148,7 +140,7 @@ private synchronized void syncStats(boolean statsEnabled) {
                     this.statsState = StatsState.MOCKED;
                 }
             } catch (Throwable t) {
-                logger.warn("failed to mock the glogue, error is {}", t);
+                logger.warn("failed to mock the glogue, error is", t);
             }
         }
     }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/executor/GraphQueryExecutor.java

@@ -185,7 +185,11 @@ public void execute(
                                                 jobName,
                                                 summary.getLogicalPlan(),
                                                 summary.getPhysicalPlan());
-                                client.submit(request, listener, timeoutConfig);
+                                client.submit(
+                                        request,
+                                        listener,
+                                        timeoutConfig,
+                                        statusCallback.getQueryLogger());
                             }
                         };
             }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/processor/IrTestOpProcessor.java

@@ -187,7 +187,8 @@ public ThrowingConsumer<Context> select(Context ctx) {
                                                         summary.getLogicalPlan(),
                                                         summary.getPhysicalPlan()),
                                                 listener,
-                                                timeoutConfig);
+                                                timeoutConfig,
+                                                statusCallback.getQueryLogger());
                                     }
                                     // request results from remote engine in a blocking way
                                     listener.request();

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/plugin/QueryStatusCallback.java

@@ -114,8 +114,6 @@ private JsonObject buildSimpleLog(boolean isSucceed, long elapsedMillis) {
     private void fillLogDetail(JsonObject logJson, String errorMsg) {
         try {
             if (this.metricsCollector.getElapsedMillis() > this.printThreshold) {
-                // todo(siyuan): the invocation of the function can cause Exception when serializing
-                // a gremlin vertex to json format
                 fillLogDetail(logJson, errorMsg, metricsCollector.getStartMillis());
             }
         } catch (Throwable t) {

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/plugin/processor/LifeCycleSupplier.java

@@ -118,7 +118,8 @@ public GremlinExecutor.LifeCycle get() {
                                                     summary.getLogicalPlan(),
                                                     summary.getPhysicalPlan()),
                                             listener,
-                                            timeoutConfig);
+                                            timeoutConfig,
+                                            statusCallback.getQueryLogger());
                                 }
                                 // request results from remote engine in a blocking way
                                 listener.request();

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/result/processor/AbstractResultProcessor.java

@@ -73,9 +73,8 @@ protected AbstractResultProcessor(
                         msg.optionalArgs(Tokens.ARGS_BATCH_SIZE)
                                 .orElse(settings.resultIterationBatchSize);
         this.resultCollectors = new ArrayList<>(this.resultCollectorsBatchSize);
-        this.responseStreamIterator =
-                new StreamIterator<>(
-                        FrontendConfig.PER_QUERY_STREAM_BUFFER_MAX_CAPACITY.get(configs));
+        int capacity = FrontendConfig.PER_QUERY_STREAM_BUFFER_MAX_CAPACITY.get(configs);
+        this.responseStreamIterator = new StreamIterator<>(capacity);
     }
 
     @Override