This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 21e6d3f7bf1 Pipe: add mark-as-general-write-request parameter in pipe
to force forwarding event (#15572)
21e6d3f7bf1 is described below
commit 21e6d3f7bf1511db98d1cb874dbd1029c1bd14f2
Author: nanxiang xia <[email protected]>
AuthorDate: Mon May 26 16:53:50 2025 +0800
Pipe: add mark-as-general-write-request parameter in pipe to force
forwarding event (#15572)
---
.../config/constant/PipeConnectorConstant.java | 6 +++++
.../pipe/connector/protocol/IoTDBConnector.java | 26 +++++++++++++++++++---
2 files changed, 29 insertions(+), 3 deletions(-)
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
index 5b8d5bc8959..5e3e9400b88 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeConnectorConstant.java
@@ -255,6 +255,12 @@ public class PipeConnectorConstant {
public static final String SINK_MARK_AS_PIPE_REQUEST_KEY =
"sink.mark-as-pipe-request";
public static final boolean CONNECTOR_MARK_AS_PIPE_REQUEST_DEFAULT_VALUE =
true;
+ public static final String CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_KEY =
+ "connector.mark-as-general-write-request";
+ public static final String SINK_MARK_AS_GENERAL_WRITE_REQUEST_KEY =
+ "sink.mark-as-general-write-request";
+ public static final boolean
CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_DEFAULT_VALUE = false;
+
public static final String CONNECTOR_SKIP_IF_KEY = "connector.skipif";
public static final String SINK_SKIP_IF_KEY = "sink.skipif";
public static final String CONNECTOR_IOTDB_SKIP_IF_NO_PRIVILEGES =
"no-privileges";
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java
index aef50daa701..8a434dc888f 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/connector/protocol/IoTDBConnector.java
@@ -103,6 +103,8 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_LOAD_TSFILE_VALIDATION_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_LOAD_TSFILE_VALIDATION_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_DEFAULT_VALUE;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_MARK_AS_PIPE_REQUEST_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_MARK_AS_PIPE_REQUEST_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.CONNECTOR_RATE_LIMIT_DEFAULT_VALUE;
@@ -132,6 +134,7 @@ import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstan
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_LOAD_BALANCE_STRATEGY_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_LOAD_TSFILE_STRATEGY_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_LOAD_TSFILE_VALIDATION_KEY;
+import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_MARK_AS_GENERAL_WRITE_REQUEST_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_MARK_AS_PIPE_REQUEST_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_RATE_LIMIT_KEY;
import static
org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant.SINK_SKIP_IF_KEY;
@@ -229,6 +232,13 @@ public abstract class IoTDBConnector implements
PipeConnector {
Arrays.asList(CONNECTOR_IOTDB_BATCH_DELAY_SECONDS_KEY,
SINK_IOTDB_BATCH_DELAY_SECONDS_KEY),
false);
+ // Check coexistence of mark-as-pipe-request and
mark-as-general-write-request
+ validator.validateSynonymAttributes(
+ Arrays.asList(CONNECTOR_MARK_AS_PIPE_REQUEST_KEY,
SINK_MARK_AS_PIPE_REQUEST_KEY),
+ Arrays.asList(
+ CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_KEY,
SINK_MARK_AS_GENERAL_WRITE_REQUEST_KEY),
+ false);
+
username =
parameters.getStringOrDefault(
Arrays.asList(
@@ -383,10 +393,20 @@ public abstract class IoTDBConnector implements
PipeConnector {
.equals(CONNECTOR_FORMAT_TS_FILE_VALUE);
LOGGER.info("IoTDBConnector isTabletBatchModeEnabled: {}",
isTabletBatchModeEnabled);
- shouldMarkAsPipeRequest =
+ final boolean shouldMarkAsGeneralWriteRequest =
parameters.getBooleanOrDefault(
- Arrays.asList(CONNECTOR_MARK_AS_PIPE_REQUEST_KEY,
SINK_MARK_AS_PIPE_REQUEST_KEY),
- CONNECTOR_MARK_AS_PIPE_REQUEST_DEFAULT_VALUE);
+ Arrays.asList(
+ CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_KEY,
+ SINK_MARK_AS_GENERAL_WRITE_REQUEST_KEY),
+ CONNECTOR_MARK_AS_GENERAL_WRITE_REQUEST_DEFAULT_VALUE);
+ if (shouldMarkAsGeneralWriteRequest) {
+ shouldMarkAsPipeRequest = false;
+ } else {
+ shouldMarkAsPipeRequest =
+ parameters.getBooleanOrDefault(
+ Arrays.asList(CONNECTOR_MARK_AS_PIPE_REQUEST_KEY,
SINK_MARK_AS_PIPE_REQUEST_KEY),
+ CONNECTOR_MARK_AS_PIPE_REQUEST_DEFAULT_VALUE);
+ }
LOGGER.info("IoTDBConnector shouldMarkAsPipeRequest: {}",
shouldMarkAsPipeRequest);
final String connectorSkipIfValue =
