This is an automated email from the ASF dual-hosted git repository. CritasWang pushed a commit to branch wx/iotdb-2.0.11-alignment in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
commit 62f4a8714f390eeb7a7635c51a417e8b866a3647 Author: CritasWang <[email protected]> AuthorDate: Thu Sep 17 17:50:47 2026 +0800 collector: fail task creation when the IoTDB source cannot consume The IoTDB push source opened its subscription on the worker thread, so a missing topic, a tsfile-format topic, an unreachable broker or a server without subscription support only produced one log line while the task stayed registered and silently never delivered. Validate the topic format and subscribe on the calling thread so start() reports the cause, and let PushSourceTask/PullSourceTask propagate a start failure even when closing the source succeeds, matching SinkTask and ProcessorTask. --- .../builtin/source/iotdb/IoTDBPushSource.java | 82 ++++++++++++++++++---- .../runtime/task/source/pull/PullSourceTask.java | 4 +- .../runtime/task/source/push/PushSourceTask.java | 4 +- 3 files changed, 75 insertions(+), 15 deletions(-) diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/iotdb/IoTDBPushSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/iotdb/IoTDBPushSource.java index eb52868..77655b1 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/iotdb/IoTDBPushSource.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/iotdb/IoTDBPushSource.java @@ -25,7 +25,10 @@ import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfi import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.rpc.subscription.config.ConsumerConstant; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; +import org.apache.iotdb.session.subscription.SubscriptionTreeSession; import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumer; +import org.apache.iotdb.session.subscription.model.Topic; import org.apache.iotdb.session.subscription.payload.SubscriptionMessage; import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType; @@ -35,6 +38,7 @@ import org.slf4j.LoggerFactory; import java.util.Iterator; import java.util.List; +import java.util.Locale; import java.util.Optional; import java.util.Properties; @@ -49,6 +53,7 @@ public class IoTDBPushSource extends PushSource { private String deviceId; private volatile boolean isStarted = true; + private SubscriptionTreePullConsumer consumer; private Thread workerThread; @Override @@ -77,29 +82,69 @@ public class IoTDBPushSource extends PushSource { @Override public void start() throws Exception { - if (workerThread == null || !workerThread.isAlive()) { - isStarted = true; - workerThread = new Thread(this::doWork); - workerThread.start(); + if (workerThread != null && workerThread.isAlive()) { + return; } - } - private void doWork() { + // Validate the topic and subscribe on the calling thread so that a missing topic, a + // tsfile-format topic, an unreachable broker or a server without subscription support fails + // task creation with the cause, instead of leaving a task that looks alive but never delivers. + requireRecordFormatTopic(); + final Properties pullProperties = new Properties(); pullProperties.put(IoTDBPushSourceConstant.HOST_KEY, host); pullProperties.put(IoTDBPushSourceConstant.PORT_KEY, port); pullProperties.put(ConsumerConstant.CONSUMER_ID_KEY, "r1"); pullProperties.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, "rg1"); - try (final SubscriptionTreePullConsumer consumer = - new SubscriptionTreePullConsumer(pullProperties)) { - consumer.open(); - consumer.subscribe(topic); + final SubscriptionTreePullConsumer pullConsumer = + new SubscriptionTreePullConsumer(pullProperties); + try { + pullConsumer.open(); + pullConsumer.subscribe(topic); + } catch (final Exception e) { + try { + pullConsumer.close(); + } catch (final Exception closeException) { + e.addSuppressed(closeException); + } + throw e; + } + + consumer = pullConsumer; + isStarted = true; + workerThread = new Thread(this::doWork, "iotdb-push-source-" + topic); + workerThread.start(); + } + + private void requireRecordFormatTopic() throws Exception { + try (final SubscriptionTreeSession session = new SubscriptionTreeSession(host, port)) { + session.open(); + final Optional<Topic> found = session.getTopic(topic); + if (!found.isPresent()) { + throw new IllegalArgumentException( + String.format( + "Topic %s does not exist on %s:%d; create it with format=%s before starting the" + + " collector IoTDB source", + topic, host, port, TopicConstant.FORMAT_RECORD_HANDLER_VALUE)); + } + final String attributes = String.valueOf(found.get().getTopicAttributes()); + if (attributes.toLowerCase(Locale.ROOT).contains("tsfilehandler")) { + throw new IllegalArgumentException( + String.format( + "Topic %s delivers tsfile messages (%s); the collector IoTDB source only consumes" + + " record-format messages, create the topic with format=%s", + topic, attributes, TopicConstant.FORMAT_RECORD_HANDLER_VALUE)); + } + } + } + private void doWork() { + try (final SubscriptionTreePullConsumer pullConsumer = consumer) { while (isStarted && !Thread.currentThread().isInterrupted()) { markPausePosition(); - final List<SubscriptionMessage> messages = consumer.poll(timeout); + final List<SubscriptionMessage> messages = pullConsumer.poll(timeout); for (final SubscriptionMessage message : messages) { final short messageType = message.getMessageType(); if (messageType == SubscriptionMessageType.RECORD_HANDLER.getType()) { @@ -109,13 +154,24 @@ public class IoTDBPushSource extends PushSource { } } else if (messageType != SubscriptionMessageType.WATERMARK.getType()) { throw new UnsupportedOperationException( - "Collector IoTDB source requires record-format subscription messages"); + String.format( + "Topic %s delivered a message of type %d; the collector IoTDB source only" + + " consumes record-format messages (format=%s)", + topic, messageType, TopicConstant.FORMAT_RECORD_HANDLER_VALUE)); } } } } catch (final Exception e) { Thread.currentThread().interrupt(); - LOGGER.error("Error in push source", e); + if (isStarted) { + LOGGER.error( + "The collector IoTDB source for topic {} stopped consuming; drop and recreate the task" + + " after fixing the cause", + topic, + e); + } else { + LOGGER.info("The collector IoTDB source for topic {} stopped", topic); + } } } diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceTask.java index 5cf7468..423595c 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceTask.java @@ -99,8 +99,10 @@ public class PullSourceTask extends SourceTask { consumers[i].consumer().close(); } catch (final Exception ex) { LOGGER.warn("Failed to close source on creation failure", ex); - throw e; } + // Like SinkTask/ProcessorTask: a source that cannot start must fail task creation + // instead of being swallowed when its cleanup succeeds. + throw e; } int finalI = i; diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java index 1c74234..ce0647b 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java @@ -81,8 +81,10 @@ public class PushSourceTask extends SourceTask { pushSources[i].close(); } catch (final Exception ex) { LOGGER.warn("Failed to close source on creation failure", ex); - throw e; } + // Like SinkTask/ProcessorTask: a source that cannot start must fail task creation + // instead of being swallowed when its cleanup succeeds. + throw e; } }
