This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rc/1.3.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/1.3.3 by this push:
new 3c02466ff39 Pipe: Fixed the bug that alter source will not alter the
reference count of ConfigRegionListeningQueue (#13626)(#13632)
3c02466ff39 is described below
commit 3c02466ff39153dd1cc8e84730f6559950e02d7a
Author: Caideyipi <[email protected]>
AuthorDate: Thu Sep 26 15:16:32 2024 +0800
Pipe: Fixed the bug that alter source will not alter the reference count of
ConfigRegionListeningQueue (#13626)(#13632)
---
.../iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java | 20 ++++++++++---
.../agent/runtime/PipeConfigNodeRuntimeAgent.java | 13 +++++----
.../agent/runtime/PipeConfigRegionListener.java | 7 ++---
.../confignode/persistence/pipe/PipeInfo.java | 33 ++++++++++++++++++----
4 files changed, 53 insertions(+), 20 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
index fb541e7dcdb..364c5d475f0 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
@@ -36,6 +36,7 @@ import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -470,7 +471,7 @@ public class IoTDBPipeAlterIT extends
AbstractPipeDualAutoIT {
try (final Connection connection = senderEnv.getConnection();
final Statement statement = connection.createStatement()) {
statement.execute(sql);
- } catch (SQLException e) {
+ } catch (final SQLException e) {
fail(e.getMessage());
}
@@ -491,12 +492,13 @@ public class IoTDBPipeAlterIT extends
AbstractPipeDualAutoIT {
TestUtils.assertDataEventuallyOnEnv(
receiverEnv, "select * from root.db.**", "Time,root.db.d1.at1,",
expectedResSet);
- // Alter pipe (modify 'source.path' and
'processor.tumbling-time.interval-seconds')
+ // Alter pipe (modify 'source.path', 'source.inclusion' and
+ // 'processor.tumbling-time.interval-seconds')
try (final Connection connection = senderEnv.getConnection();
final Statement statement = connection.createStatement()) {
statement.execute(
- "alter pipe a2b modify source('source' =
'iotdb-source','source.path'='root.db.d2.**') modify processor
('processor.tumbling-time.interval-seconds'='2')");
- } catch (SQLException e) {
+ "alter pipe a2b modify source('source' =
'iotdb-source','source.path'='root.db.d2.**', 'source.inclusion'='all') modify
processor ('processor.tumbling-time.interval-seconds'='2')");
+ } catch (final SQLException e) {
fail(e.getMessage());
}
@@ -527,5 +529,15 @@ public class IoTDBPipeAlterIT extends
AbstractPipeDualAutoIT {
"select * from root.db.** where time > 10000",
"Time,root.db.d1.at1,root.db.d2.at1,",
expectedResSet);
+
+ // Create database on sender
+ if (!TestUtils.tryExecuteNonQueryWithRetry(
+ senderEnv, "create timeSeries root.db.d2.at2 int32")) {
+ fail();
+ }
+
+ // Check database on receiver
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv, "count timeSeries", "count(timeseries),",
Collections.singleton("3,"));
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
index c3262c68181..cc6056522ef 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java
@@ -35,7 +35,6 @@ import
org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
public class PipeConfigNodeRuntimeAgent implements IService {
@@ -90,12 +89,13 @@ public class PipeConfigNodeRuntimeAgent implements IService
{
return regionListener.listener();
}
- public void increaseListenerReference(PipeParameters parameters) throws
IllegalPathException {
+ public void increaseListenerReference(final PipeParameters parameters)
+ throws IllegalPathException {
regionListener.increaseReference(parameters);
}
- public void decreaseListenerReference(PipeParameters parameters)
- throws IllegalPathException, IOException {
+ public void decreaseListenerReference(final PipeParameters parameters)
+ throws IllegalPathException {
regionListener.decreaseReference(parameters);
}
@@ -120,7 +120,7 @@ public class PipeConfigNodeRuntimeAgent implements IService
{
//////////////////////////// Runtime Exception Handlers
////////////////////////////
- public void report(EnrichedEvent event, PipeRuntimeException
pipeRuntimeException) {
+ public void report(final EnrichedEvent event, final PipeRuntimeException
pipeRuntimeException) {
if (event.getPipeTaskMeta() != null) {
report(event.getPipeTaskMeta(), pipeRuntimeException);
} else {
@@ -128,7 +128,8 @@ public class PipeConfigNodeRuntimeAgent implements IService
{
}
}
- private void report(PipeTaskMeta pipeTaskMeta, PipeRuntimeException
pipeRuntimeException) {
+ private void report(
+ final PipeTaskMeta pipeTaskMeta, final PipeRuntimeException
pipeRuntimeException) {
LOGGER.warn(
"Report PipeRuntimeException to local PipeTaskMeta({}), exception
message: {}",
pipeTaskMeta,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java
index 458ed1c57a8..7c8bcca9f36 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java
@@ -24,7 +24,6 @@ import
org.apache.iotdb.confignode.manager.pipe.extractor.ConfigRegionListeningF
import
org.apache.iotdb.confignode.manager.pipe.extractor.ConfigRegionListeningQueue;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
public class PipeConfigRegionListener {
@@ -38,7 +37,7 @@ public class PipeConfigRegionListener {
return listeningQueue;
}
- public synchronized void increaseReference(PipeParameters parameters)
+ public synchronized void increaseReference(final PipeParameters parameters)
throws IllegalPathException {
if
(!ConfigRegionListeningFilter.parseListeningPlanTypeSet(parameters).isEmpty()) {
listeningQueueReferenceCount++;
@@ -48,8 +47,8 @@ public class PipeConfigRegionListener {
}
}
- public synchronized void decreaseReference(PipeParameters parameters)
- throws IllegalPathException, IOException {
+ public synchronized void decreaseReference(final PipeParameters parameters)
+ throws IllegalPathException {
if
(!ConfigRegionListeningFilter.parseListeningPlanTypeSet(parameters).isEmpty()) {
listeningQueueReferenceCount--;
if (listeningQueueReferenceCount == 0) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java
index fff58164626..b80070c1d21 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeInfo.java
@@ -159,14 +159,35 @@ public class PipeInfo implements SnapshotProcessor {
public TSStatus alterPipe(final AlterPipePlanV2 plan) {
try {
+ final Optional<PipeMeta> pipeMetaBeforeAlter =
+ Optional.ofNullable(
+
pipeTaskInfo.getPipeMetaByPipeName(plan.getPipeStaticMeta().getPipeName()));
+
pipeTaskInfo.alterPipe(plan);
- PipeConfigNodeAgent.task()
- .handleSinglePipeMetaChanges(
-
pipeTaskInfo.getPipeMetaByPipeName(plan.getPipeStaticMeta().getPipeName()));
- PipeTemporaryMetaMetrics.getInstance()
- .handleTemporaryMetaChanges(pipeTaskInfo.getPipeMetaList());
- return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ final TPushPipeMetaRespExceptionMessage message =
+ PipeConfigNodeAgent.task()
+ .handleSinglePipeMetaChanges(
+
pipeTaskInfo.getPipeMetaByPipeName(plan.getPipeStaticMeta().getPipeName()));
+ if (message == null) {
+ pipeMetaBeforeAlter.ifPresent(
+ meta -> {
+ try {
+ PipeConfigNodeAgent.runtime()
+
.decreaseListenerReference(meta.getStaticMeta().getExtractorParameters());
+ } catch (final Exception e) {
+ throw new PipeException("Failed to decrease listener
reference", e);
+ }
+ });
+ PipeConfigNodeAgent.runtime()
+
.increaseListenerReference(plan.getPipeStaticMeta().getExtractorParameters());
+ PipeTemporaryMetaMetrics.getInstance()
+ .handleTemporaryMetaChanges(pipeTaskInfo.getPipeMetaList());
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ } else {
+ return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
+ .setMessage(message.getMessage());
+ }
} catch (final Exception e) {
LOGGER.error("Failed to alter pipe", e);
return new TSStatus(TSStatusCode.PIPE_ERROR.getStatusCode())
