You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/08/28 14:33:13 UTC
[iotdb] branch pipe-he-increase-1.2 created (now 64f22072462)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch pipe-he-increase-1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 64f22072462 Pipe: Increase the injection frequency of HeartBeatEvent to reduce the delay in log transferring (#10970)
This branch includes the following new commits:
new 64f22072462 Pipe: Increase the injection frequency of HeartBeatEvent to reduce the delay in log transferring (#10970)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: Pipe: Increase the injection frequency of HeartBeatEvent to reduce the delay in log transferring (#10970)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch pipe-he-increase-1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 64f22072462af057a5698dd9c6b5da5ef02d0e3e
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Mon Aug 28 18:03:53 2023 +0800
Pipe: Increase the injection frequency of HeartBeatEvent to reduce the delay in log transferring (#10970)
(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,