This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new 9f31e15a3d5 Pipe: Increase the injection frequency of HeartBeatEvent
to reduce the delay in log transferring (#10970) (#10978)
9f31e15a3d5 is described below
commit 9f31e15a3d59f80345c0ad9cada255e5ee5da2da
Author: Steve Yurong Su <[email protected]>
AuthorDate: Mon Aug 28 23:24:51 2023 +0800
Pipe: Increase the injection frequency of HeartBeatEvent to reduce the
delay in log transferring (#10970) (#10978)
(cherry picked from commit daebc419e0ecab1b8d60cfcc82d688a843cd6b75)
---
iotdb-connector/flink-sql-iotdb-connector/pom.xml | 2 +-
.../pipe/agent/runtime/PipeCronEventInjector.java | 70 ++++++++++++++++++++++
.../db/pipe/agent/runtime/PipeRuntimeAgent.java | 7 ++-
.../iotdb/db/pipe/agent/task/PipeTaskAgent.java | 2 +-
.../config/constant/PipeConnectorConstant.java | 2 +-
.../event/common/heartbeat/PipeHeartbeatEvent.java | 12 ++--
.../event/realtime/PipeRealtimeEventFactory.java | 6 +-
.../listener/PipeInsertionDataNodeListener.java | 6 +-
.../iotdb/commons/concurrent/ThreadName.java | 2 +
9 files changed, 97 insertions(+), 12 deletions(-)
diff --git a/iotdb-connector/flink-sql-iotdb-connector/pom.xml
b/iotdb-connector/flink-sql-iotdb-connector/pom.xml
index 85657b35701..dd4c49ee66d 100644
--- a/iotdb-connector/flink-sql-iotdb-connector/pom.xml
+++ b/iotdb-connector/flink-sql-iotdb-connector/pom.xml
@@ -29,7 +29,7 @@
</parent>
<artifactId>flink-sql-iotdb-connector</artifactId>
<version>1.2.0-SNAPSHOT</version>
- <name>IoTDB: Connector: Apache Flink SQL Connector</name>
+ <name>IoTDB: Connector: Apache Flink SQL</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>1.17.0</flink.version>
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeCronEventInjector.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeCronEventInjector.java
new file mode 100644
index 00000000000..8ec9798e187
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeCronEventInjector.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.runtime;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import
org.apache.iotdb.db.pipe.extractor.realtime.listener.PipeInsertionDataNodeListener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class PipeCronEventInjector {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PipeCronEventInjector.class);
+
+ private static final int CRON_EVENT_INJECTOR_INTERVAL_SECONDS = 1;
+
+ private static final ScheduledExecutorService CRON_EVENT_INJECTOR_EXECUTOR =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.PIPE_RUNTIME_CRON_EVENT_INJECTOR.getName());
+
+ private Future<?> injectorFuture;
+
+ public synchronized void start() {
+ if (injectorFuture == null) {
+ injectorFuture =
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ CRON_EVENT_INJECTOR_EXECUTOR,
+ this::inject,
+ CRON_EVENT_INJECTOR_INTERVAL_SECONDS,
+ CRON_EVENT_INJECTOR_INTERVAL_SECONDS,
+ TimeUnit.SECONDS);
+ LOGGER.info("Pipe cron event injector is started successfully.");
+ }
+ }
+
+ private synchronized void inject() {
+ PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(false);
+ }
+
+ public synchronized void stop() {
+ if (injectorFuture != null) {
+ injectorFuture.cancel(false);
+ injectorFuture = null;
+ LOGGER.info("Pipe cron event injector is stopped successfully.");
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
index 0f91b346611..dacd51d4684 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeRuntimeAgent.java
@@ -43,7 +43,9 @@ public class PipeRuntimeAgent implements IService {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeRuntimeAgent.class);
private static final int DATA_NODE_ID =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
- private static final AtomicBoolean isShutdown = new AtomicBoolean(false);
+ private final AtomicBoolean isShutdown = new AtomicBoolean(false);
+
+ private final PipeCronEventInjector pipeCronEventInjector = new
PipeCronEventInjector();
private final SimpleConsensusProgressIndexAssigner
simpleConsensusProgressIndexAssigner =
new SimpleConsensusProgressIndexAssigner();
@@ -66,6 +68,7 @@ public class PipeRuntimeAgent implements IService {
public synchronized void start() throws StartupException {
PipeConfig.getInstance().printAllConfigs();
PipeAgentLauncher.launchPipeTaskAgent();
+ pipeCronEventInjector.start();
isShutdown.set(false);
}
@@ -77,6 +80,7 @@ public class PipeRuntimeAgent implements IService {
}
isShutdown.set(true);
+ pipeCronEventInjector.stop();
PipeAgent.task().dropAllPipeTasks();
}
@@ -96,6 +100,7 @@ public class PipeRuntimeAgent implements IService {
}
////////////////////// Recover ProgressIndex Assigner //////////////////////
+
public void assignRecoverProgressIndexForTsFileRecovery(TsFileResource
tsFileResource) {
tsFileResource.recoverProgressIndex(
new RecoverProgressIndex(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
index 660b9d962e4..e54203f0f7c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeTaskAgent.java
@@ -843,6 +843,6 @@ public class PipeTaskAgent {
}
resp.setPipeMetaList(pipeMetaBinaryList);
- PipeInsertionDataNodeListener.getInstance().listenToHeartbeat();
+ PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true);
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
index 3c870a9ba15..2a5b8e0a47a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeConnectorConstant.java
@@ -33,7 +33,7 @@ public class PipeConnectorConstant {
public static final boolean CONNECTOR_IOTDB_BATCH_MODE_ENABLED_DEFAULT_VALUE
= true;
public static final String CONNECTOR_IOTDB_BATCH_DELAY_KEY =
"connector.batch.max-delay-seconds";
- public static final int CONNECTOR_IOTDB_BATCH_DELAY_DEFAULT_VALUE = 10;
+ public static final int CONNECTOR_IOTDB_BATCH_DELAY_DEFAULT_VALUE = 1;
public static final String CONNECTOR_IOTDB_BATCH_SIZE_KEY =
"connector.batch.size-bytes";
public static final long CONNECTOR_IOTDB_BATCH_SIZE_DEFAULT_VALUE = 16 * MB;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
index f194182b832..71d65a6b7cd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
@@ -40,15 +40,19 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
private long timeProcessed;
private long timeTransferred;
- public PipeHeartbeatEvent(String dataRegionId) {
+ private final boolean shouldPrintMessage;
+
+ public PipeHeartbeatEvent(String dataRegionId, boolean shouldPrintMessage) {
super(null, null);
this.dataRegionId = dataRegionId;
+ this.shouldPrintMessage = shouldPrintMessage;
}
- public PipeHeartbeatEvent(String dataRegionId, long timePublished) {
+ public PipeHeartbeatEvent(String dataRegionId, long timePublished, boolean
shouldPrintMessage) {
super(null, null);
this.dataRegionId = dataRegionId;
this.timePublished = timePublished;
+ this.shouldPrintMessage = shouldPrintMessage;
}
@Override
@@ -60,7 +64,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
public boolean internallyDecreaseResourceReferenceCount(String
holderMessage) {
// PipeName == null indicates that the event is the raw event at disruptor,
// not the event copied and passed to the extractor
- if (pipeName != null && LOGGER.isInfoEnabled()) {
+ if (shouldPrintMessage && pipeName != null && LOGGER.isInfoEnabled()) {
LOGGER.info(this.toString());
}
return true;
@@ -74,7 +78,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
@Override
public EnrichedEvent shallowCopySelfAndBindPipeTaskMetaForProgressReport(
PipeTaskMeta pipeTaskMeta, String pattern) {
- return new PipeHeartbeatEvent(dataRegionId, timePublished);
+ return new PipeHeartbeatEvent(dataRegionId, timePublished,
shouldPrintMessage);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
index 72c15279878..3a95f9ccf26 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
@@ -49,8 +49,10 @@ public class PipeRealtimeEventFactory {
resource);
}
- public static PipeRealtimeEvent createRealtimeEvent(String dataRegionId) {
- return new PipeRealtimeEvent(new PipeHeartbeatEvent(dataRegionId), null,
null, null);
+ public static PipeRealtimeEvent createRealtimeEvent(
+ String dataRegionId, boolean shouldPrintMessage) {
+ return new PipeRealtimeEvent(
+ new PipeHeartbeatEvent(dataRegionId, shouldPrintMessage), null, null,
null);
}
private PipeRealtimeEventFactory() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java
index 7446bef72ac..e3c7a62b6d9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java
@@ -128,13 +128,15 @@ public class PipeInsertionDataNodeListener {
PipeRealtimeEventFactory.createRealtimeEvent(walEntryHandler,
insertNode, tsFileResource));
}
- public void listenToHeartbeat() {
+ public void listenToHeartbeat(boolean shouldPrintMessage) {
if (listenToInsertNodeExtractorCount.get() == 0 &&
listenToTsFileExtractorCount.get() == 0) {
return;
}
dataRegionId2Assigner.forEach(
- (key, value) ->
value.publishToAssign(PipeRealtimeEventFactory.createRealtimeEvent(key)));
+ (key, value) ->
+ value.publishToAssign(
+ PipeRealtimeEventFactory.createRealtimeEvent(key,
shouldPrintMessage)));
}
/////////////////////////////// singleton ///////////////////////////////
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
index 024d1c4386c..e1cb4a3ff1b 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
@@ -131,6 +131,7 @@ public enum ThreadName {
PIPE_RUNTIME_META_SYNCER("Pipe-Runtime-Meta-Syncer"),
PIPE_RUNTIME_HEARTBEAT("Pipe-Runtime-Heartbeat"),
PIPE_RUNTIME_PROCEDURE_SUBMITTER("Pipe-Runtime-Procedure-Submitter"),
+ PIPE_RUNTIME_CRON_EVENT_INJECTOR("Pipe-Runtime-Cron-Event-Injector"),
PIPE_ASYNC_CONNECTOR_CLIENT_POOL("Pipe-Async-Connector-Client-Pool"),
PIPE_WAL_RESOURCE_TTL_CHECKER("Pipe-WAL-Resource-TTL-Checker"),
PIPE_RECEIVER_AIR_GAP_AGENT("Pipe-Receiver-Air-Gap-Agent"),
@@ -267,6 +268,7 @@ public enum ThreadName {
PIPE_RUNTIME_META_SYNCER,
PIPE_RUNTIME_HEARTBEAT,
PIPE_RUNTIME_PROCEDURE_SUBMITTER,
+ PIPE_RUNTIME_CRON_EVENT_INJECTOR,
PIPE_ASYNC_CONNECTOR_CLIENT_POOL,
PIPE_WAL_RESOURCE_TTL_CHECKER,
PIPE_RECEIVER_AIR_GAP_AGENT,
