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 2025/03/05 09:32:54 UTC
(iotdb-extras) branch polish-collector updated (ce8819e -> 1ed399e)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch polish-collector
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
from ce8819e 1
new cc61e06 refactor
new 9e25f77 refactor
new 1ed399e refactor
The 3 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.
Summary of changes:
.../collector/plugin/api/CollectorPullSource.java | 22 ----
.../api/{CollectorSource.java => PullSource.java} | 2 +-
.../{CollectorPushSource.java => PushSource.java} | 11 +-
.../collector/plugin/api/StoppablePlugin.java | 24 ----
.../CollectorProcessorRuntimeConfiguration.java} | 45 ++-----
.../CollectorRuntimeEnvironment.java} | 64 +++++----
.../CollectorSinkRuntimeConfiguration.java} | 45 ++-----
.../CollectorSourceRuntimeConfiguration.java} | 45 ++-----
.../api/event/CollectorEvent.java} | 15 +--
.../SourceEvent.java => api/event/DemoEvent.java} | 21 ++-
.../plugin/{ => builtin}/BuiltinPlugin.java | 38 ++----
.../processor/DoNothingProcessor.java | 36 +++---
.../sink/DemoSink.java} | 51 +++++---
.../{ => builtin}/source/HttpPullSource.java | 17 +--
.../{ => builtin}/source/HttpPushSource.java | 24 +---
.../collector/runtime/plugin/PluginFactory.java | 73 -----------
.../collector/runtime/plugin/PluginRuntime.java | 45 +++++++
.../plugin/constructor/PluginConstructor.java | 99 ++++++++++++++
.../plugin/constructor/ProcessorConstructor.java | 48 +++++++
.../plugin/constructor/SinkConstructor.java | 51 ++++++++
.../plugin/constructor/SourceConstructor.java | 59 +++++++++
.../collector/runtime/plugin/meta/PluginMeta.java | 119 +++++++++++++++++
.../runtime/plugin/meta/PluginMetaKeeper.java | 101 +++++++++++++++
.../collector/runtime/task/{def => }/Task.java | 73 ++++++-----
.../runtime/task/{def => }/TaskCombiner.java | 54 +++-----
.../iotdb/collector/runtime/task/TaskRuntime.java | 110 ++++++++--------
.../runtime/task/def/processor/ProcessorTask.java | 119 -----------------
.../runtime/task/def/sink/SinkConsumer.java | 23 ----
.../collector/runtime/task/def/sink/SinkTask.java | 104 ---------------
.../runtime/task/def/source/PullSourceTask.java | 82 ------------
.../runtime/task/def/source/PushSourceTask.java | 73 -----------
.../runtime/task/def/source/SourceTask.java | 132 -------------------
.../task/{execution => event}/EventCollector.java | 10 +-
.../task/{execution => event}/EventContainer.java | 2 +-
.../runtime/task/execution/EventConsumer.java | 78 -----------
.../runtime/task/processor/ProcessorConsumer.java | 57 ++++++++
.../ProcessorExceptionHandler.java} | 18 ++-
.../runtime/task/processor/ProcessorTask.java | 125 ++++++++++++++++++
.../collector/runtime/task/sink/SinkConsumer.java | 54 ++++++++
.../SinkExceptionHandler.java} | 19 ++-
.../collector/runtime/task/sink/SinkTask.java | 120 +++++++++++++++++
.../collector/runtime/task/source/SourceTask.java | 64 +++++++++
.../task/source/pull/PullSourceConsumer.java} | 24 ++--
.../runtime/task/source/pull/PullSourceTask.java | 144 +++++++++++++++++++++
.../runtime/task/source/push/PushSourceTask.java | 101 +++++++++++++++
45 files changed, 1504 insertions(+), 1137 deletions(-)
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPullSource.java
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/{CollectorSource.java => PullSource.java} (92%)
copy iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/{CollectorPushSource.java => PushSource.java} (82%)
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/StoppablePlugin.java
copy iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/{RuntimeConfig.java => customizer/CollectorProcessorRuntimeConfiguration.java} (56%)
copy iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/{RuntimeConfig.java => customizer/CollectorRuntimeEnvironment.java} (53%)
copy iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/{RuntimeConfig.java => customizer/CollectorSinkRuntimeConfiguration.java} (56%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/{RuntimeConfig.java => customizer/CollectorSourceRuntimeConfiguration.java} (56%)
copy iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/{runtime/task/execution/EventContainer.java => plugin/api/event/CollectorEvent.java} (77%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/{event/SourceEvent.java => api/event/DemoEvent.java} (71%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/{ => builtin}/BuiltinPlugin.java (57%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/{ => builtin}/processor/DoNothingProcessor.java (63%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/{sink/SessionSink.java => builtin/sink/DemoSink.java} (56%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/{ => builtin}/source/HttpPullSource.java (80%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/{ => builtin}/source/HttpPushSource.java (80%)
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginFactory.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/PluginConstructor.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMeta.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMetaKeeper.java
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/{def => }/Task.java (54%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/{def => }/TaskCombiner.java (58%)
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/processor/ProcessorTask.java
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkConsumer.java
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkTask.java
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PullSourceTask.java
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PushSourceTask.java
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/SourceTask.java
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/{execution => event}/EventCollector.java (83%)
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/{execution => event}/EventContainer.java (94%)
delete mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumer.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorConsumer.java
copy iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/{execution/EventConsumerExceptionHandler.java => processor/ProcessorExceptionHandler.java} (68%)
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkConsumer.java
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/{execution/EventConsumerExceptionHandler.java => sink/SinkExceptionHandler.java} (68%)
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java
rename iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/{plugin/api/CollectorPushSource.java => runtime/task/source/pull/PullSourceConsumer.java} (60%)
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceTask.java
create mode 100644 iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java
(iotdb-extras) 02/03: refactor
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch polish-collector
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
commit 9e25f77b1a3730e77da06fca790a197105292d95
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Mar 5 17:15:52 2025 +0800
refactor
---
.../iotdb/collector/plugin/BuiltinPlugin.java | 28 +----
.../collector/plugin/api/CollectorPullSource.java | 22 ----
.../api/{CollectorSource.java => PullSource.java} | 2 +-
.../{CollectorPushSource.java => PushSource.java} | 11 +-
.../iotdb/collector/plugin/api/RuntimeConfig.java | 63 -----------
.../collector/plugin/api/StoppablePlugin.java | 24 -----
.../collector/plugin/source/HttpPullSource.java | 7 +-
.../collector/plugin/source/HttpPushSource.java | 14 +--
.../collector/runtime/plugin/PluginFactory.java | 73 -------------
.../collector/runtime/plugin/PluginRuntime.java | 45 ++++++++
.../plugin/constructor/PluginConstructor.java | 99 +++++++++++++++++
.../plugin/constructor/ProcessorConstructor.java | 48 +++++++++
.../plugin/constructor/SinkConstructor.java | 51 +++++++++
.../plugin/constructor/SourceConstructor.java | 59 ++++++++++
.../collector/runtime/plugin/meta/PluginMeta.java | 119 +++++++++++++++++++++
.../runtime/plugin/meta/PluginMetaKeeper.java | 101 +++++++++++++++++
.../apache/iotdb/collector/runtime/task/Task.java | 42 +++-----
.../iotdb/collector/runtime/task/TaskCombiner.java | 52 ++++-----
.../iotdb/collector/runtime/task/TaskRuntime.java | 97 +++++++++--------
.../runtime/task/event/EventCollector.java | 3 +
.../runtime/task/processor/ProcessorTask.java | 42 ++++----
.../collector/runtime/task/sink/SinkTask.java | 36 +++++--
.../collector/runtime/task/source/SourceTask.java | 30 ++++--
.../task/source/pull/PullSourceConsumer.java | 8 +-
.../runtime/task/source/pull/PullSourceTask.java | 61 +++++++----
.../runtime/task/source/push/PushSourceTask.java | 81 +++++++++-----
26 files changed, 802 insertions(+), 416 deletions(-)
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java
index 683a2c4..789d2f3 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java
@@ -24,24 +24,19 @@ import org.apache.iotdb.collector.plugin.sink.SessionSink;
import org.apache.iotdb.collector.plugin.source.HttpPullSource;
import org.apache.iotdb.collector.plugin.source.HttpPushSource;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Set;
-
public enum BuiltinPlugin {
- // PushSources
+ // Push Sources
HTTP_PUSH_SOURCE("http-push-source", HttpPushSource.class),
- // PullSources
+ // Pull Sources
HTTP_PULL_SOURCE("http-pull-source", HttpPullSource.class),
// Processors
DO_NOTHING_PROCESSOR("do-nothing-processor", DoNothingProcessor.class),
// Sinks
- IOTDB_SESSION_SINK("iotdb-session-sink", SessionSink.class);
+ IOTDB_THRIFT_SINK("iotdb-thrift-sink", SessionSink.class);
private final String collectorPluginName;
private final Class<?> collectorPluginClass;
@@ -53,28 +48,15 @@ public enum BuiltinPlugin {
this.className = collectorPluginClass.getName();
}
- public String getCollectorPluginName() {
+ public String getPluginName() {
return collectorPluginName;
}
- public Class<?> getCollectorPluginClass() {
+ public Class<?> getPluginClass() {
return collectorPluginClass;
}
public String getClassName() {
return className;
}
-
- public static final Set<String> SHOW_COLLECTOR_PLUGINS_BLACKLIST =
- Collections.unmodifiableSet(
- new HashSet<>(
- Arrays.asList(
- // PushSources
- HTTP_PUSH_SOURCE.getCollectorPluginName().toUpperCase(),
- // PullSources
- HTTP_PULL_SOURCE.getCollectorPluginName().toUpperCase(),
- // Processors
- DO_NOTHING_PROCESSOR.getCollectorPluginName().toUpperCase(),
- // Sinks
- IOTDB_SESSION_SINK.getCollectorPluginName().toUpperCase())));
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPullSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPullSource.java
deleted file mode 100644
index 097111a..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPullSource.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.collector.plugin.api;
-
-public abstract class CollectorPullSource implements CollectorSource {}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/PullSource.java
similarity index 92%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorSource.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/PullSource.java
index 2e483e1..e15a493 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorSource.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/PullSource.java
@@ -21,4 +21,4 @@ package org.apache.iotdb.collector.plugin.api;
import org.apache.iotdb.pipe.api.PipeSource;
-public interface CollectorSource extends StoppablePlugin, PipeSource {}
+public abstract class PullSource implements PipeSource {}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPushSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/PushSource.java
similarity index 82%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPushSource.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/PushSource.java
index 622ed52..e8eebf4 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPushSource.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/PushSource.java
@@ -19,14 +19,19 @@
package org.apache.iotdb.collector.plugin.api;
+import org.apache.iotdb.pipe.api.PipeSource;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.event.Event;
-public abstract class CollectorPushSource implements CollectorSource {
+public abstract class PushSource implements PipeSource {
- protected final EventCollector collector;
+ protected EventCollector collector;
- public CollectorPushSource(final EventCollector collector) {
+ public PushSource() {
+ this.collector = null;
+ }
+
+ public final void setCollector(final EventCollector collector) {
this.collector = collector;
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/RuntimeConfig.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/RuntimeConfig.java
deleted file mode 100644
index 1da84bc..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/RuntimeConfig.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.collector.plugin.api;
-
-import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
-import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment;
-
-public class RuntimeConfig implements PipeExtractorRuntimeConfiguration {
-
- public static class RuntimeEnvironment implements PipeRuntimeEnvironment {
-
- public int getParallelism() {
- return 0;
- }
-
- public int getParallelismIndex() {
- return 0;
- }
-
- @Override
- public String getPipeName() {
- return "";
- }
-
- @Override
- public long getCreationTime() {
- return 0;
- }
- }
-
- @Override
- public PipeRuntimeEnvironment getRuntimeEnvironment() {
- return new PipeRuntimeEnvironment() {
-
- @Override
- public String getPipeName() {
- return "";
- }
-
- @Override
- public long getCreationTime() {
- return 0;
- }
- };
- }
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/StoppablePlugin.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/StoppablePlugin.java
deleted file mode 100644
index 8a97564..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/StoppablePlugin.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.collector.plugin.api;
-
-public interface StoppablePlugin {
- void stop() throws Exception;
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java
index 83881a3..c809b5f 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.collector.plugin.source;
-import org.apache.iotdb.collector.plugin.api.CollectorPullSource;
+import org.apache.iotdb.collector.plugin.api.PullSource;
import org.apache.iotdb.collector.plugin.event.SourceEvent;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration;
@@ -34,7 +34,7 @@ import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
-public class HttpPullSource extends CollectorPullSource {
+public class HttpPullSource extends PullSource {
private static final Logger LOGGER = LoggerFactory.getLogger(HttpPullSource.class);
@@ -62,9 +62,6 @@ public class HttpPullSource extends CollectorPullSource {
return event;
}
- @Override
- public void stop() throws Exception {}
-
@Override
public void close() {}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java
index 96d03d0..7ad2cc3 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java
@@ -19,9 +19,8 @@
package org.apache.iotdb.collector.plugin.source;
-import org.apache.iotdb.collector.plugin.api.CollectorPushSource;
+import org.apache.iotdb.collector.plugin.api.PushSource;
import org.apache.iotdb.collector.plugin.event.SourceEvent;
-import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -34,17 +33,13 @@ import org.slf4j.LoggerFactory;
import java.util.Random;
import java.util.concurrent.TimeUnit;
-public class HttpPushSource extends CollectorPushSource {
+public class HttpPushSource extends PushSource {
private static final Logger LOGGER = LoggerFactory.getLogger(HttpPushSource.class);
private volatile boolean isStarted = true;
private Thread workerThread;
- public HttpPushSource(final EventCollector collector) {
- super(collector);
- }
-
@Override
public void validate(final PipeParameterValidator validator) {}
@@ -80,11 +75,6 @@ public class HttpPushSource extends CollectorPushSource {
}
}
- @Override
- public void stop() throws Exception {
- this.isStarted = false;
- }
-
@Override
public void close() {
isStarted = false;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginFactory.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginFactory.java
deleted file mode 100644
index 351771a..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginFactory.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.collector.runtime.plugin;
-
-import org.apache.iotdb.collector.plugin.BuiltinPlugin;
-import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor;
-import org.apache.iotdb.collector.plugin.sink.SessionSink;
-import org.apache.iotdb.collector.plugin.source.HttpPullSource;
-import org.apache.iotdb.pipe.api.PipePlugin;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.Supplier;
-
-public class PluginFactory {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(PluginFactory.class);
-
- protected final Map<String, Supplier<PipePlugin>> pluginConstructors = new HashMap<>();
-
- public PluginFactory() {
- initFactory();
- }
-
- private void initFactory() {
- pluginConstructors.put(
- BuiltinPlugin.HTTP_PULL_SOURCE.getCollectorPluginName(), HttpPullSource::new);
- pluginConstructors.put(
- BuiltinPlugin.DO_NOTHING_PROCESSOR.getCollectorPluginName(), DoNothingProcessor::new);
- pluginConstructors.put(
- BuiltinPlugin.IOTDB_SESSION_SINK.getCollectorPluginName(), SessionSink::new);
- LOGGER.info("builtin plugin has been initialized");
- }
-
- public static <T> T createInstance(final Class<T> clazz) {
- try {
- final Constructor<T> constructor = clazz.getDeclaredConstructor();
- constructor.setAccessible(true);
- return constructor.newInstance();
- } catch (final NoSuchMethodException e) {
- LOGGER.warn("class {} is abstract class.", clazz, e);
- } catch (final IllegalAccessException e) {
- LOGGER.warn("failed to visit class {} constructor method.", clazz, e);
- } catch (final InstantiationException e) {
- LOGGER.warn("failed to instantiate class {}.", clazz, e);
- } catch (final InvocationTargetException e) {
- LOGGER.warn("the constructor threw an exception.", e);
- }
- throw new RuntimeException();
- }
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginRuntime.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginRuntime.java
index 6b220d1..adbad7d 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginRuntime.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginRuntime.java
@@ -19,8 +19,53 @@
package org.apache.iotdb.collector.runtime.plugin;
+import org.apache.iotdb.collector.runtime.plugin.constructor.ProcessorConstructor;
+import org.apache.iotdb.collector.runtime.plugin.constructor.SinkConstructor;
+import org.apache.iotdb.collector.runtime.plugin.constructor.SourceConstructor;
+import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
+import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.PipeSink;
+import org.apache.iotdb.pipe.api.PipeSource;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
public class PluginRuntime implements AutoCloseable {
+ private final PluginMetaKeeper metaKeeper;
+ private final SourceConstructor sourceConstructor;
+ private final ProcessorConstructor processorConstructor;
+ private final SinkConstructor sinkConstructor;
+
+ public PluginRuntime() {
+ this.metaKeeper = new PluginMetaKeeper();
+ this.sourceConstructor = new SourceConstructor(metaKeeper);
+ this.processorConstructor = new ProcessorConstructor(metaKeeper);
+ this.sinkConstructor = new SinkConstructor(metaKeeper);
+ }
+
+ public PipeSource constructSource(final PipeParameters sourceParameters) {
+ return sourceConstructor.reflectPlugin(sourceParameters);
+ }
+
+ public boolean isPullSource(final PipeParameters sourceParameters) throws Exception {
+ try (final PipeSource source = constructSource(sourceParameters)) {
+ return sourceConstructor.isPullSource(source);
+ }
+ }
+
+ public boolean isPushSource(final PipeParameters sourceParameters) throws Exception {
+ try (final PipeSource source = constructSource(sourceParameters)) {
+ return sourceConstructor.isPushSource(source);
+ }
+ }
+
+ public PipeProcessor constructProcessor(final PipeParameters processorParameters) {
+ return processorConstructor.reflectPlugin(processorParameters);
+ }
+
+ public PipeSink constructSink(final PipeParameters sinkParameters) {
+ return sinkConstructor.reflectPlugin(sinkParameters);
+ }
+
public boolean createPlugin() {
return true;
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/PluginConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/PluginConstructor.java
new file mode 100644
index 0000000..c64ef39
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/PluginConstructor.java
@@ -0,0 +1,99 @@
+/*
+ * 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.collector.runtime.plugin.constructor;
+
+import org.apache.iotdb.collector.runtime.plugin.meta.PluginMeta;
+import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
+import org.apache.iotdb.pipe.api.PipePlugin;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+
+public abstract class PluginConstructor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PluginConstructor.class);
+
+ private final PluginMetaKeeper pluginMetaKeeper;
+
+ protected final Map<String, Supplier<PipePlugin>> pluginConstructors = new HashMap<>();
+
+ protected PluginConstructor(PluginMetaKeeper pluginMetaKeeper) {
+ this.pluginMetaKeeper = pluginMetaKeeper;
+ initConstructors();
+ }
+
+ // New plugins shall be put here
+ protected abstract void initConstructors();
+
+ public abstract PipePlugin reflectPlugin(PipeParameters pipeParameters);
+
+ public PipePlugin reflectPluginByKey(String pluginKey) {
+ return pluginConstructors.getOrDefault(pluginKey, () -> reflect(pluginKey)).get();
+ }
+
+ private PipePlugin reflect(String pluginName) {
+ if (pluginMetaKeeper == null) {
+ throw new PipeException(
+ "Failed to reflect PipePlugin instance, because PluginMetaKeeper is null.");
+ }
+
+ if (pluginName == null) {
+ throw new PipeException(
+ "Failed to reflect PipePlugin instance, because plugin name is null.");
+ }
+
+ final PluginMeta information = pluginMetaKeeper.getPipePluginMeta(pluginName);
+ if (information == null) {
+ String errorMessage =
+ String.format(
+ "Failed to reflect PipePlugin instance, because "
+ + "PipePlugin %s has not been registered.",
+ pluginName.toUpperCase());
+ LOGGER.warn(errorMessage);
+ throw new PipeException(errorMessage);
+ }
+
+ try {
+ final Class<?> pluginClass =
+ information.isBuiltin()
+ ? pluginMetaKeeper.getBuiltinPluginClass(information.getPluginName())
+ : Class.forName(information.getClassName()); // TODO
+ return (PipePlugin) pluginClass.getDeclaredConstructor().newInstance();
+ } catch (InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException e) {
+ String errorMessage =
+ String.format(
+ "Failed to reflect PipePlugin %s(%s) instance, because %s",
+ pluginName, information.getClassName(), e);
+ LOGGER.warn(errorMessage, e);
+ throw new PipeException(errorMessage);
+ }
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java
new file mode 100644
index 0000000..7c3bdc0
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java
@@ -0,0 +1,48 @@
+/*
+ * 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.collector.runtime.plugin.constructor;
+
+import org.apache.iotdb.collector.plugin.BuiltinPlugin;
+import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor;
+import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
+import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+public class ProcessorConstructor extends PluginConstructor {
+
+ public ProcessorConstructor(PluginMetaKeeper pluginMetaKeeper) {
+ super(pluginMetaKeeper);
+ }
+
+ @Override
+ protected void initConstructors() {
+ pluginConstructors.put(
+ BuiltinPlugin.DO_NOTHING_PROCESSOR.getPluginName(), DoNothingProcessor::new);
+ }
+
+ @Override
+ public final PipeProcessor reflectPlugin(PipeParameters processorParameters) {
+ return (PipeProcessor)
+ reflectPluginByKey(
+ processorParameters
+ .getStringOrDefault("processor", BuiltinPlugin.DO_NOTHING_PROCESSOR.getPluginName())
+ .toLowerCase());
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java
new file mode 100644
index 0000000..dc640ac
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java
@@ -0,0 +1,51 @@
+/*
+ * 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.collector.runtime.plugin.constructor;
+
+import org.apache.iotdb.collector.plugin.BuiltinPlugin;
+import org.apache.iotdb.collector.plugin.sink.SessionSink;
+import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
+import org.apache.iotdb.pipe.api.PipeSink;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+public class SinkConstructor extends PluginConstructor {
+
+ public SinkConstructor(PluginMetaKeeper pluginMetaKeeper) {
+ super(pluginMetaKeeper);
+ }
+
+ @Override
+ protected void initConstructors() {
+ pluginConstructors.put(BuiltinPlugin.IOTDB_THRIFT_SINK.getPluginName(), SessionSink::new);
+ }
+
+ @Override
+ public final PipeSink reflectPlugin(PipeParameters sinkParameters) {
+ if (sinkParameters.hasAttribute("sink")) {
+ throw new IllegalArgumentException("sink attribute is required");
+ }
+
+ return (PipeSink)
+ reflectPluginByKey(
+ sinkParameters
+ .getStringOrDefault("sink", BuiltinPlugin.IOTDB_THRIFT_SINK.getPluginName())
+ .toLowerCase());
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java
new file mode 100644
index 0000000..871dc00
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java
@@ -0,0 +1,59 @@
+/*
+ * 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.collector.runtime.plugin.constructor;
+
+import org.apache.iotdb.collector.plugin.BuiltinPlugin;
+import org.apache.iotdb.collector.plugin.api.PullSource;
+import org.apache.iotdb.collector.plugin.api.PushSource;
+import org.apache.iotdb.collector.plugin.source.HttpPullSource;
+import org.apache.iotdb.collector.plugin.source.HttpPushSource;
+import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
+import org.apache.iotdb.pipe.api.PipeSource;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+
+public class SourceConstructor extends PluginConstructor {
+
+ public SourceConstructor(PluginMetaKeeper pluginMetaKeeper) {
+ super(pluginMetaKeeper);
+ }
+
+ @Override
+ protected void initConstructors() {
+ pluginConstructors.put(BuiltinPlugin.HTTP_PULL_SOURCE.getPluginName(), HttpPullSource::new);
+ pluginConstructors.put(BuiltinPlugin.HTTP_PUSH_SOURCE.getPluginName(), HttpPushSource::new);
+ }
+
+ @Override
+ public final PipeSource reflectPlugin(PipeParameters sourceParameters) {
+ if (sourceParameters.hasAttribute("source")) {
+ throw new IllegalArgumentException("source attribute is required");
+ }
+
+ return (PipeSource) reflectPluginByKey(sourceParameters.getString("source").toLowerCase());
+ }
+
+ public boolean isPullSource(PipeSource source) {
+ return source instanceof PullSource;
+ }
+
+ public boolean isPushSource(PipeSource source) {
+ return source instanceof PushSource;
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMeta.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMeta.java
new file mode 100644
index 0000000..2c9b6cc
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMeta.java
@@ -0,0 +1,119 @@
+/*
+ * 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.collector.runtime.plugin.meta;
+
+import java.util.Objects;
+
+public class PluginMeta {
+
+ private final String pluginName;
+ private final String className;
+
+ // jarName and jarMD5 are used to identify the jar file.
+ // they could be null if the plugin is built-in. they should be both null or both not null.
+ private final boolean isBuiltin;
+ private final String jarName;
+ private final String jarMD5;
+
+ public PluginMeta(
+ String pluginName, String className, boolean isBuiltin, String jarName, String jarMD5) {
+ this.pluginName = Objects.requireNonNull(pluginName).toUpperCase();
+ this.className = Objects.requireNonNull(className);
+
+ this.isBuiltin = isBuiltin;
+ if (isBuiltin) {
+ this.jarName = jarName;
+ this.jarMD5 = jarMD5;
+ } else {
+ this.jarName = Objects.requireNonNull(jarName);
+ this.jarMD5 = Objects.requireNonNull(jarMD5);
+ }
+ }
+
+ public PluginMeta(String pluginName, String className) {
+ this.pluginName = Objects.requireNonNull(pluginName).toUpperCase();
+ this.className = Objects.requireNonNull(className);
+
+ this.isBuiltin = true;
+ this.jarName = null;
+ this.jarMD5 = null;
+ }
+
+ public boolean isBuiltin() {
+ return isBuiltin;
+ }
+
+ public String getPluginName() {
+ return pluginName;
+ }
+
+ public String getClassName() {
+ return className;
+ }
+
+ public String getJarName() {
+ return jarName;
+ }
+
+ public String getJarMD5() {
+ return jarMD5;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ PluginMeta that = (PluginMeta) obj;
+ return pluginName.equals(that.pluginName)
+ && className.equals(that.className)
+ && isBuiltin == that.isBuiltin
+ && Objects.equals(jarName, that.jarName)
+ && Objects.equals(jarMD5, that.jarMD5);
+ }
+
+ @Override
+ public int hashCode() {
+ return pluginName.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return "PluginMeta{"
+ + "pluginName='"
+ + pluginName
+ + '\''
+ + ", className='"
+ + className
+ + '\''
+ + ", isBuiltin="
+ + isBuiltin
+ + ", jarName='"
+ + jarName
+ + '\''
+ + ", jarMD5='"
+ + jarMD5
+ + '\''
+ + '}';
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMetaKeeper.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMetaKeeper.java
new file mode 100644
index 0000000..4d9ddfd
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/meta/PluginMetaKeeper.java
@@ -0,0 +1,101 @@
+/*
+ * 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.collector.runtime.plugin.meta;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class PluginMetaKeeper {
+
+ protected final Map<String, PluginMeta> pipePluginNameToMetaMap = new ConcurrentHashMap<>();
+ protected final Map<String, Class<?>> builtinPipePluginNameToClassMap = new ConcurrentHashMap<>();
+
+ public PluginMetaKeeper() {
+ loadBuiltinPlugins();
+ }
+
+ private void loadBuiltinPlugins() {
+ // for (final BuiltinPipePlugin builtinPipePlugin : BuiltinPipePlugin.values()) {
+ // final String pipePluginName = builtinPipePlugin.getPipePluginName();
+ // final Class<?> pipePluginClass = builtinPipePlugin.getPipePluginClass();
+ // final String className = builtinPipePlugin.getClassName();
+ //
+ // addPipePluginMeta(pipePluginName, new PluginMeta(pipePluginName, className));
+ // addBuiltinPluginClass(pipePluginName, pipePluginClass);
+ // addPipePluginVisibility(
+ // pipePluginName, VisibilityUtils.calculateFromPluginClass(pipePluginClass));
+ // }
+ }
+
+ public void addPipePluginMeta(String pluginName, PluginMeta pluginMeta) {
+ pipePluginNameToMetaMap.put(pluginName.toUpperCase(), pluginMeta);
+ }
+
+ public void removePipePluginMeta(String pluginName) {
+ pipePluginNameToMetaMap.remove(pluginName.toUpperCase());
+ }
+
+ public PluginMeta getPipePluginMeta(String pluginName) {
+ return pipePluginNameToMetaMap.get(pluginName.toUpperCase());
+ }
+
+ public Iterable<PluginMeta> getAllPipePluginMeta() {
+ return pipePluginNameToMetaMap.values();
+ }
+
+ public boolean containsPipePlugin(String pluginName) {
+ return pipePluginNameToMetaMap.containsKey(pluginName.toUpperCase());
+ }
+
+ private void addBuiltinPluginClass(String pluginName, Class<?> builtinPipePluginClass) {
+ builtinPipePluginNameToClassMap.put(pluginName.toUpperCase(), builtinPipePluginClass);
+ }
+
+ public Class<?> getBuiltinPluginClass(String pluginName) {
+ return builtinPipePluginNameToClassMap.get(pluginName.toUpperCase());
+ }
+
+ public String getPluginNameByJarName(String jarName) {
+ for (Map.Entry<String, PluginMeta> entry : pipePluginNameToMetaMap.entrySet()) {
+ if (jarName.equals(entry.getValue().getJarName())) {
+ return entry.getKey();
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PluginMetaKeeper that = (PluginMetaKeeper) o;
+ return pipePluginNameToMetaMap.equals(that.pipePluginNameToMetaMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(pipePluginNameToMetaMap);
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/Task.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/Task.java
index 610e8ae..2b27001 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/Task.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/Task.java
@@ -66,47 +66,31 @@ public abstract class Task {
}
}
- public final synchronized void create() {
- try {
- resume();
- createInternal();
- } catch (final Exception e) {
- LOGGER.warn("Failed to create task", e);
- }
+ public final synchronized void create() throws Exception {
+ resume();
+ createInternal();
}
public abstract void createInternal() throws Exception;
- public final synchronized void start() {
- try {
- resume();
- startInternal();
- } catch (final Exception e) {
- LOGGER.warn("Failed to start task", e);
- }
+ public final synchronized void start() throws Exception {
+ resume();
+ startInternal();
}
public abstract void startInternal() throws Exception;
- public final synchronized void stop() {
- try {
- pause();
- stopInternal();
- } catch (final Exception e) {
- LOGGER.warn("Failed to stop task", e);
- }
+ public final synchronized void stop() throws Exception {
+ pause();
+ stopInternal();
}
public abstract void stopInternal() throws Exception;
- public final synchronized void drop() {
- try {
- pause();
- isDropped.set(true);
- dropInternal();
- } catch (final Exception e) {
- LOGGER.warn("Failed to drop task", e);
- }
+ public final synchronized void drop() throws Exception {
+ pause();
+ isDropped.set(true);
+ dropInternal();
}
public abstract void dropInternal() throws Exception;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskCombiner.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskCombiner.java
index b100160..c835bc9 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskCombiner.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskCombiner.java
@@ -36,45 +36,29 @@ public class TaskCombiner {
this.sink = sink;
}
- // Disruptor consumers must be started before producers
- public void create() {
- try {
- sink.create();
- processor.create();
- source.create();
- } catch (final Exception e) {
- LOGGER.warn("Failed to create task", e);
- }
+ public void create() throws Exception {
+ sink.create();
+ processor.create();
+ source.create();
}
- // Disruptor consumers must be started before producers
- public void start() {
- try {
- sink.start();
- processor.start();
- source.start();
- } catch (final Exception e) {
- LOGGER.warn("Failed to start task", e);
- }
+ public void start() throws Exception {
+ sink.start();
+ processor.start();
+ source.start();
}
- public void stop() {
- try {
- source.stop();
- processor.stop();
- sink.stop();
- } catch (final Exception e) {
- LOGGER.warn("Failed to stop task", e);
- }
+ public void stop() throws Exception {
+ source.stop();
+ processor.stop();
+ sink.stop();
}
- public void drop() {
- try {
- source.drop();
- processor.drop();
- sink.drop();
- } catch (final Exception e) {
- LOGGER.warn("Failed to drop task", e);
- }
+ public void drop() throws Exception {
+ stop();
+
+ source.drop();
+ processor.drop();
+ sink.drop();
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java
index 178a3ae..ec3a4f1 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java
@@ -37,13 +37,13 @@ public class TaskRuntime implements AutoCloseable {
private final Map<String, TaskCombiner> tasks = new ConcurrentHashMap<>();
- public Response createTask(
+ public synchronized Response createTask(
final String taskId,
final Map<String, String> sourceAttribute,
final Map<String, String> processorAttribute,
final Map<String, String> sinkAttribute) {
try {
- if (validateTaskIsExist(taskId)) {
+ if (tasks.containsKey(taskId)) {
return Response.status(Response.Status.CONFLICT)
.entity(String.format("task %s has existed", taskId))
.build();
@@ -52,89 +52,98 @@ public class TaskRuntime implements AutoCloseable {
final SinkTask sinkTask = new SinkTask(taskId, sinkAttribute);
final ProcessorTask processorTask =
new ProcessorTask(taskId, processorAttribute, sinkTask.makeProducer());
- final SourceTask sourceTask = SourceTask.construct(taskId, sourceAttribute, processorTask);
- final TaskCombiner taskRepository = new TaskCombiner(sourceTask, processorTask, sinkTask);
+ final SourceTask sourceTask =
+ SourceTask.construct(taskId, sourceAttribute, processorTask.makeProducer());
- tasks.put(taskId, taskRepository);
- taskRepository.create();
+ final TaskCombiner taskCombiner = new TaskCombiner(sourceTask, processorTask, sinkTask);
+ tasks.put(taskId, taskCombiner);
+ taskCombiner.create();
LOGGER.info("Successfully created task {}", taskId);
return Response.status(Response.Status.CREATED)
.entity(String.format("Successfully created task %s", taskId))
.build();
} catch (final Exception e) {
- LOGGER.warn("Failed to create task", e);
+ LOGGER.warn("Failed to create task {} because {}", taskId, e.getMessage(), e);
return Response.serverError()
.entity(String.format("Failed to create task %s, because %s", taskId, e.getMessage()))
.build();
}
}
- public boolean alterTask() {
- return true;
- }
-
- public Response startTask(final String taskId) {
- if (!validateTaskIsExist(taskId)) {
+ public synchronized Response startTask(final String taskId) {
+ if (!tasks.containsKey(taskId)) {
return Response.status(Response.Status.NOT_FOUND)
.entity(String.format("task %s not found", taskId))
.build();
}
- tasks.get(taskId).start();
- LOGGER.info("Task {} started successfully", taskId);
- return Response.status(Response.Status.OK)
- .entity(String.format("task %s start successfully", taskId))
- .build();
+ try {
+ tasks.get(taskId).start();
+
+ LOGGER.info("Task {} start successfully", taskId);
+ return Response.status(Response.Status.OK)
+ .entity(String.format("task %s start successfully", taskId))
+ .build();
+ } catch (Exception e) {
+ LOGGER.warn("Failed to start task {} because {}", taskId, e.getMessage(), e);
+ return Response.serverError()
+ .entity(String.format("Failed to start task %s, because %s", taskId, e.getMessage()))
+ .build();
+ }
}
- public Response stopTask(final String taskId) {
- if (!validateTaskIsExist(taskId)) {
+ public synchronized Response stopTask(final String taskId) {
+ if (!tasks.containsKey(taskId)) {
return Response.status(Response.Status.NOT_FOUND)
.entity(String.format("task %s not found", taskId))
.build();
}
try {
- final TaskCombiner taskRepository = tasks.get(taskId);
- if (taskRepository != null) {
- taskRepository.stop();
- }
+ tasks.get(taskId).stop();
+
+ LOGGER.info("Task {} stop successfully", taskId);
+ return Response.status(Response.Status.OK)
+ .entity(String.format("task %s stop successfully", taskId))
+ .build();
} catch (final Exception e) {
- LOGGER.warn("Failed to stop task", e);
+ LOGGER.warn("Failed to stop task {} because {}", taskId, e.getMessage(), e);
return Response.serverError()
.entity(String.format("Failed to stop task %s, because %s", taskId, e.getMessage()))
.build();
}
-
- LOGGER.info("Task {} stopped successfully", taskId);
- return Response.status(Response.Status.OK)
- .entity(String.format("task %s stop successfully", taskId))
- .build();
}
- public Response dropTask(final String taskId) {
- if (!validateTaskIsExist(taskId)) {
+ public synchronized Response dropTask(final String taskId) {
+ if (!tasks.containsKey(taskId)) {
return Response.status(Response.Status.NOT_FOUND)
.entity(String.format("task %s not found", taskId))
.build();
}
- tasks.remove(taskId).drop();
- LOGGER.info("Task {} dropped successfully", taskId);
- return Response.status(Response.Status.OK)
- .entity(String.format("task %s drop successfully", taskId))
- .build();
- }
+ try {
+ tasks.remove(taskId).drop();
- private boolean validateTaskIsExist(final String taskId) {
- if (tasks.containsKey(taskId)) {
- return true;
+ LOGGER.info("Task {} drop successfully", taskId);
+ return Response.status(Response.Status.OK)
+ .entity(String.format("task %s drop successfully", taskId))
+ .build();
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to drop task {} because {}", taskId, e.getMessage(), e);
+ return Response.serverError()
+ .entity(String.format("Failed to drop task %s, because %s", taskId, e.getMessage()))
+ .build();
}
- LOGGER.warn("Task {} not found", taskId);
- return false;
}
@Override
- public void close() throws Exception {}
+ public synchronized void close() throws Exception {
+ final long currentTime = System.currentTimeMillis();
+ for (final TaskCombiner taskCombiner : tasks.values()) {
+ taskCombiner.drop();
+ }
+ tasks.clear();
+ LOGGER.info("Task runtime closed in {}ms", System.currentTimeMillis() - currentTime);
+ }
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventCollector.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventCollector.java
index 2e2675a..a96b61a 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventCollector.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventCollector.java
@@ -23,6 +23,9 @@ import org.apache.iotdb.pipe.api.event.Event;
import com.lmax.disruptor.RingBuffer;
+import javax.annotation.concurrent.ThreadSafe;
+
+@ThreadSafe
public class EventCollector implements org.apache.iotdb.pipe.api.collector.EventCollector {
private final RingBuffer<EventContainer> ringBuffer;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java
index 51def10..e23524d 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java
@@ -20,13 +20,12 @@
package org.apache.iotdb.collector.runtime.task.processor;
import org.apache.iotdb.collector.plugin.api.customizer.CollectorProcessorRuntimeConfiguration;
-import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor;
-import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
+import org.apache.iotdb.collector.runtime.plugin.PluginRuntime;
import org.apache.iotdb.collector.runtime.task.Task;
import org.apache.iotdb.collector.runtime.task.event.EventCollector;
import org.apache.iotdb.collector.runtime.task.event.EventContainer;
+import org.apache.iotdb.collector.service.RuntimeService;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
-import org.apache.iotdb.pipe.api.event.Event;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
@@ -63,19 +62,32 @@ public class ProcessorTask extends Task {
@Override
public void createInternal() throws Exception {
+ final PluginRuntime pluginRuntime =
+ RuntimeService.plugin().isPresent() ? RuntimeService.plugin().get() : null;
+ if (pluginRuntime == null) {
+ throw new IllegalStateException("Plugin runtime is down");
+ }
+
final long creationTime = System.currentTimeMillis();
processorConsumers = new ProcessorConsumer[parallelism];
for (int i = 0; i < parallelism; i++) {
- // TODO: PluginFactory
processorConsumers[i] =
- new ProcessorConsumer(
- PluginFactory.createInstance(DoNothingProcessor.class), sinkProducer);
- processorConsumers[i].consumer().validate(new PipeParameterValidator(parameters));
- processorConsumers[i]
- .consumer()
- .customize(
- parameters,
- new CollectorProcessorRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ new ProcessorConsumer(pluginRuntime.constructProcessor(parameters), sinkProducer);
+ try {
+ processorConsumers[i].consumer().validate(new PipeParameterValidator(parameters));
+ processorConsumers[i]
+ .consumer()
+ .customize(
+ parameters,
+ new CollectorProcessorRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ } catch (final Exception e) {
+ try {
+ processorConsumers[i].consumer().close();
+ } catch (final Exception ex) {
+ LOGGER.warn("Failed to close sink on creation failure", ex);
+ }
+ throw e;
+ }
}
disruptor.handleEventsWithWorkerPool(processorConsumers);
@@ -110,10 +122,4 @@ public class ProcessorTask extends Task {
public EventCollector makeProducer() {
return new EventCollector(disruptor.getRingBuffer());
}
-
- public void publishEvent(final Event event) {
- disruptor
- .getRingBuffer()
- .publishEvent((container, sequence, o) -> container.setEvent(event), event);
- }
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java
index fbda5a4..17c7052 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java
@@ -20,11 +20,11 @@
package org.apache.iotdb.collector.runtime.task.sink;
import org.apache.iotdb.collector.plugin.api.customizer.CollectorSinkRuntimeConfiguration;
-import org.apache.iotdb.collector.plugin.sink.SessionSink;
-import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
+import org.apache.iotdb.collector.runtime.plugin.PluginRuntime;
import org.apache.iotdb.collector.runtime.task.Task;
import org.apache.iotdb.collector.runtime.task.event.EventCollector;
import org.apache.iotdb.collector.runtime.task.event.EventContainer;
+import org.apache.iotdb.collector.service.RuntimeService;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import com.lmax.disruptor.BlockingWaitStrategy;
@@ -57,18 +57,32 @@ public class SinkTask extends Task {
@Override
public void createInternal() throws Exception {
+ final PluginRuntime pluginRuntime =
+ RuntimeService.plugin().isPresent() ? RuntimeService.plugin().get() : null;
+ if (pluginRuntime == null) {
+ throw new IllegalStateException("Plugin runtime is down");
+ }
+
final long creationTime = System.currentTimeMillis();
consumers = new SinkConsumer[parallelism];
for (int i = 0; i < parallelism; i++) {
- // TODO: PluginFactory
- consumers[i] = new SinkConsumer(PluginFactory.createInstance(SessionSink.class));
- consumers[i].consumer().validate(new PipeParameterValidator(parameters));
- consumers[i]
- .consumer()
- .customize(
- parameters,
- new CollectorSinkRuntimeConfiguration(taskId, creationTime, parallelism, i));
- consumers[i].consumer().handshake();
+ consumers[i] = new SinkConsumer(pluginRuntime.constructSink(parameters));
+ try {
+ consumers[i].consumer().validate(new PipeParameterValidator(parameters));
+ consumers[i]
+ .consumer()
+ .customize(
+ parameters,
+ new CollectorSinkRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ consumers[i].consumer().handshake();
+ } catch (final Exception e) {
+ try {
+ consumers[i].consumer().close();
+ } catch (final Exception ex) {
+ LOGGER.warn("Failed to close sink on creation failure", ex);
+ }
+ throw e;
+ }
}
disruptor.handleEventsWithWorkerPool(consumers);
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java
index 3b91cfa..6386615 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java
@@ -19,28 +19,46 @@
package org.apache.iotdb.collector.runtime.task.source;
+import org.apache.iotdb.collector.runtime.plugin.PluginRuntime;
import org.apache.iotdb.collector.runtime.task.Task;
-import org.apache.iotdb.collector.runtime.task.processor.ProcessorTask;
+import org.apache.iotdb.collector.runtime.task.event.EventCollector;
import org.apache.iotdb.collector.runtime.task.source.pull.PullSourceTask;
+import org.apache.iotdb.collector.runtime.task.source.push.PushSourceTask;
+import org.apache.iotdb.collector.service.RuntimeService;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import java.util.Map;
public abstract class SourceTask extends Task {
- protected final ProcessorTask processorTask;
+ protected final EventCollector processorProducer;
protected SourceTask(
final String taskId,
final Map<String, String> attributes,
- final ProcessorTask processorTask) {
+ final EventCollector processorProducer) {
super(taskId, attributes);
- this.processorTask = processorTask;
+ this.processorProducer = processorProducer;
}
public static SourceTask construct(
final String taskId,
final Map<String, String> attributes,
- final ProcessorTask processorTask) {
- return new PullSourceTask(taskId, attributes, processorTask);
+ final EventCollector processorProducer)
+ throws Exception {
+ final PluginRuntime pluginRuntime =
+ RuntimeService.plugin().isPresent() ? RuntimeService.plugin().get() : null;
+ if (pluginRuntime == null) {
+ throw new IllegalStateException("Plugin runtime is down");
+ }
+
+ final PipeParameters parameters = new PipeParameters(attributes);
+ if (pluginRuntime.isPullSource(parameters)) {
+ return new PullSourceTask(taskId, attributes, processorProducer);
+ }
+ if (pluginRuntime.isPushSource(parameters)) {
+ return new PushSourceTask(taskId, attributes, processorProducer);
+ }
+ throw new IllegalArgumentException("Unsupported source type");
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceConsumer.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceConsumer.java
index d4f56f7..85baedf 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceConsumer.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceConsumer.java
@@ -19,20 +19,20 @@
package org.apache.iotdb.collector.runtime.task.source.pull;
-import org.apache.iotdb.collector.plugin.api.CollectorPullSource;
+import org.apache.iotdb.collector.plugin.api.PullSource;
import org.apache.iotdb.pipe.api.collector.EventCollector;
class PullSourceConsumer {
- private final CollectorPullSource pullSource;
+ private final PullSource pullSource;
private final EventCollector eventCollector;
- PullSourceConsumer(final CollectorPullSource pullSource, final EventCollector eventCollector) {
+ PullSourceConsumer(final PullSource pullSource, final EventCollector eventCollector) {
this.pullSource = pullSource;
this.eventCollector = eventCollector;
}
- CollectorPullSource consumer() {
+ PullSource consumer() {
return pullSource;
}
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 bec9e75..b47054a 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
@@ -19,12 +19,12 @@
package org.apache.iotdb.collector.runtime.task.source.pull;
+import org.apache.iotdb.collector.plugin.api.PullSource;
import org.apache.iotdb.collector.plugin.api.customizer.CollectorSourceRuntimeConfiguration;
-import org.apache.iotdb.collector.plugin.source.HttpPullSource;
-import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
+import org.apache.iotdb.collector.runtime.plugin.PluginRuntime;
import org.apache.iotdb.collector.runtime.task.event.EventCollector;
-import org.apache.iotdb.collector.runtime.task.processor.ProcessorTask;
import org.apache.iotdb.collector.runtime.task.source.SourceTask;
+import org.apache.iotdb.collector.service.RuntimeService;
import org.apache.iotdb.commons.concurrent.IoTThreadFactory;
import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -45,19 +45,23 @@ public class PullSourceTask extends SourceTask {
private static final Map<String, ExecutorService> REGISTERED_EXECUTOR_SERVICES =
new ConcurrentHashMap<>();
- private final EventCollector processorProducer;
private PullSourceConsumer[] consumers;
public PullSourceTask(
final String taskId,
- final Map<String, String> sourceParams,
- final ProcessorTask processorTask) {
- super(taskId, sourceParams, processorTask);
- processorProducer = processorTask.makeProducer();
+ final Map<String, String> attributes,
+ final EventCollector processorProducer) {
+ super(taskId, attributes, processorProducer);
}
@Override
public void createInternal() throws Exception {
+ final PluginRuntime pluginRuntime =
+ RuntimeService.plugin().isPresent() ? RuntimeService.plugin().get() : null;
+ if (pluginRuntime == null) {
+ throw new IllegalStateException("Plugin runtime is down");
+ }
+
REGISTERED_EXECUTOR_SERVICES.putIfAbsent(
taskId,
new WrappedThreadPoolExecutor(
@@ -74,14 +78,23 @@ public class PullSourceTask extends SourceTask {
for (int i = 0; i < parallelism; i++) {
consumers[i] =
new PullSourceConsumer(
- PluginFactory.createInstance(HttpPullSource.class), processorProducer);
- consumers[i].consumer().validate(new PipeParameterValidator(parameters));
- consumers[i]
- .consumer()
- .customize(
- parameters,
- new CollectorSourceRuntimeConfiguration(taskId, creationTime, parallelism, i));
- consumers[i].consumer().start();
+ (PullSource) pluginRuntime.constructSource(parameters), processorProducer);
+ try {
+ consumers[i].consumer().validate(new PipeParameterValidator(parameters));
+ consumers[i]
+ .consumer()
+ .customize(
+ parameters,
+ new CollectorSourceRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ consumers[i].consumer().start();
+ } catch (final Exception e) {
+ try {
+ consumers[i].consumer().close();
+ } catch (final Exception ex) {
+ LOGGER.warn("Failed to close source on creation failure", ex);
+ throw e;
+ }
+ }
int finalI = i;
REGISTERED_EXECUTOR_SERVICES
@@ -102,17 +115,27 @@ public class PullSourceTask extends SourceTask {
}
@Override
- public void startInternal() throws Exception {
+ public void startInternal() {
// do nothing
}
@Override
- public void stopInternal() throws Exception {
+ public void stopInternal() {
// do nothing
}
@Override
- public void dropInternal() throws Exception {
+ public void dropInternal() {
+ if (consumers != null) {
+ for (int i = 0; i < parallelism; i++) {
+ try {
+ consumers[i].consumer().close();
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to close source", e);
+ }
+ }
+ }
+
final ExecutorService executorService = REGISTERED_EXECUTOR_SERVICES.remove(taskId);
if (executorService != null) {
executorService.shutdown();
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 0ea4e16..938f29f 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
@@ -19,10 +19,13 @@
package org.apache.iotdb.collector.runtime.task.source.push;
-import org.apache.iotdb.collector.plugin.source.HttpPushSource;
+import org.apache.iotdb.collector.plugin.api.PushSource;
+import org.apache.iotdb.collector.plugin.api.customizer.CollectorSourceRuntimeConfiguration;
+import org.apache.iotdb.collector.runtime.plugin.PluginRuntime;
import org.apache.iotdb.collector.runtime.task.event.EventCollector;
-import org.apache.iotdb.collector.runtime.task.processor.ProcessorTask;
import org.apache.iotdb.collector.runtime.task.source.SourceTask;
+import org.apache.iotdb.collector.service.RuntimeService;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,37 +36,65 @@ public class PushSourceTask extends SourceTask {
private static final Logger LOGGER = LoggerFactory.getLogger(PushSourceTask.class);
- private final EventCollector collector;
+ private PushSource[] pushSources;
public PushSourceTask(
final String taskId,
final Map<String, String> sourceParams,
- final ProcessorTask processorTask) {
- super(taskId, sourceParams, processorTask);
+ final EventCollector processorProducer) {
+ super(taskId, sourceParams, processorProducer);
+ }
+
+ @Override
+ public void createInternal() throws Exception {
+ final PluginRuntime pluginRuntime =
+ RuntimeService.plugin().isPresent() ? RuntimeService.plugin().get() : null;
+ if (pluginRuntime == null) {
+ throw new IllegalStateException("Plugin runtime is down");
+ }
- this.collector = processorTask.makeProducer();
+ final long creationTime = System.currentTimeMillis();
+ pushSources = new PushSource[parallelism];
+ for (int i = 0; i < parallelism; i++) {
+ pushSources[i] = (PushSource) pluginRuntime.constructSource(parameters);
+ pushSources[i].setCollector(processorProducer);
+
+ try {
+ pushSources[i].validate(new PipeParameterValidator(parameters));
+ pushSources[i].customize(
+ parameters,
+ new CollectorSourceRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ pushSources[i].start();
+ } catch (final Exception e) {
+ try {
+ pushSources[i].close();
+ } catch (final Exception ex) {
+ LOGGER.warn("Failed to close source on creation failure", ex);
+ throw e;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void startInternal() {
+ // do nothing
+ }
+
+ @Override
+ public void stopInternal() {
+ // do nothing
}
@Override
- public void createInternal() {
- createSourceTask();
- for (int i = 0; i < sourceParallelismNum; i++) {
- // use sourceAttribute later
- try (final HttpPushSource source = new HttpPushSource(collector)) {
- addSourceTask(source);
- SOURCE_EXECUTOR_SERVICE
- .get(taskId)
- .submit(
- () -> {
- try {
- source.start();
- } catch (final Exception e) {
- LOGGER.warn("Failed to start push source", e);
- throw new RuntimeException(e);
- }
- });
- } catch (Exception e) {
- LOGGER.warn("failed to create instance of HttpSource", e);
+ public void dropInternal() {
+ if (pushSources != null) {
+ for (int i = 0; i < parallelism; i++) {
+ try {
+ pushSources[i].close();
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to close source", e);
+ }
}
}
}
(iotdb-extras) 03/03: refactor
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch polish-collector
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
commit 1ed399e355b4872d220aa9b60945822d56385761
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Mar 5 17:32:45 2025 +0800
refactor
---
.../event/CollectorEvent.java} | 25 ++---------
.../SourceEvent.java => api/event/DemoEvent.java} | 21 ++++-----
.../plugin/{ => builtin}/BuiltinPlugin.java | 12 ++---
.../processor/DoNothingProcessor.java | 36 +++++++--------
.../sink/DemoSink.java} | 51 ++++++++++++++--------
.../{ => builtin}/source/HttpPullSource.java | 10 ++---
.../{ => builtin}/source/HttpPushSource.java | 10 ++---
.../plugin/constructor/ProcessorConstructor.java | 4 +-
.../plugin/constructor/SinkConstructor.java | 6 +--
.../plugin/constructor/SourceConstructor.java | 6 +--
.../runtime/task/sink/SinkExceptionHandler.java | 1 +
11 files changed, 88 insertions(+), 94 deletions(-)
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/event/CollectorEvent.java
similarity index 67%
copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java
copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/event/CollectorEvent.java
index 0c097ce..7fa251e 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/event/CollectorEvent.java
@@ -17,29 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.collector.plugin.event;
+package org.apache.iotdb.collector.plugin.api.event;
import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
-public class SourceEvent implements Event {
- private String name;
+public abstract class CollectorEvent implements Event {
- public SourceEvent() {}
-
- public SourceEvent(final String name) {
- this.name = name;
- }
-
- public String getName() {
- return name;
- }
-
- public void setName(final String name) {
- this.name = name;
- }
-
- @Override
- public String toString() {
- return "SourceEvent [name=" + name + "]";
- }
+ public abstract TabletInsertionEvent toTabletInsertionEvent();
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/event/DemoEvent.java
similarity index 71%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/event/DemoEvent.java
index 0c097ce..451ee88 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/event/DemoEvent.java
@@ -17,29 +17,24 @@
* under the License.
*/
-package org.apache.iotdb.collector.plugin.event;
+package org.apache.iotdb.collector.plugin.api.event;
import org.apache.iotdb.pipe.api.event.Event;
-public class SourceEvent implements Event {
- private String name;
+public class DemoEvent implements Event {
- public SourceEvent() {}
+ private final String value;
- public SourceEvent(final String name) {
- this.name = name;
+ public DemoEvent(final String value) {
+ this.value = value;
}
- public String getName() {
- return name;
- }
-
- public void setName(final String name) {
- this.name = name;
+ public String getValue() {
+ return value;
}
@Override
public String toString() {
- return "SourceEvent [name=" + name + "]";
+ return "DemoEvent [value = " + value + "]";
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/BuiltinPlugin.java
similarity index 81%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/BuiltinPlugin.java
index 789d2f3..203570c 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/BuiltinPlugin.java
@@ -17,12 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.collector.plugin;
+package org.apache.iotdb.collector.plugin.builtin;
-import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor;
-import org.apache.iotdb.collector.plugin.sink.SessionSink;
-import org.apache.iotdb.collector.plugin.source.HttpPullSource;
-import org.apache.iotdb.collector.plugin.source.HttpPushSource;
+import org.apache.iotdb.collector.plugin.builtin.processor.DoNothingProcessor;
+import org.apache.iotdb.collector.plugin.builtin.sink.DemoSink;
+import org.apache.iotdb.collector.plugin.builtin.source.HttpPullSource;
+import org.apache.iotdb.collector.plugin.builtin.source.HttpPushSource;
public enum BuiltinPlugin {
@@ -36,7 +36,7 @@ public enum BuiltinPlugin {
DO_NOTHING_PROCESSOR("do-nothing-processor", DoNothingProcessor.class),
// Sinks
- IOTDB_THRIFT_SINK("iotdb-thrift-sink", SessionSink.class);
+ IOTDB_THRIFT_SINK("iotdb-thrift-sink", DemoSink.class);
private final String collectorPluginName;
private final Class<?> collectorPluginClass;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/processor/DoNothingProcessor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/processor/DoNothingProcessor.java
similarity index 63%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/processor/DoNothingProcessor.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/processor/DoNothingProcessor.java
index 2bf2ee9..163bf0a 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/processor/DoNothingProcessor.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/processor/DoNothingProcessor.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.collector.plugin.processor;
+package org.apache.iotdb.collector.plugin.builtin.processor;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.collector.EventCollector;
@@ -26,42 +26,42 @@ import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.io.IOException;
public class DoNothingProcessor implements PipeProcessor {
- private static final Logger LOGGER = LoggerFactory.getLogger(DoNothingProcessor.class);
-
@Override
- public void validate(final PipeParameterValidator validator) throws Exception {
- // Do Nothing
+ public void validate(PipeParameterValidator validator) {
+ // do nothing
}
@Override
public void customize(
- final PipeParameters parameters, final PipeProcessorRuntimeConfiguration configuration)
- throws Exception {
- // Do Nothing
+ PipeParameters parameters, PipeProcessorRuntimeConfiguration configuration) {
+ // do nothing
}
@Override
- public void process(
- final TabletInsertionEvent tabletInsertionEvent, final EventCollector eventCollector)
- throws Exception {
- LOGGER.info("DoNothingProcessor process tabletInsertionEvent: {}", tabletInsertionEvent);
+ public void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector)
+ throws IOException {
eventCollector.collect(tabletInsertionEvent);
}
@Override
- public void process(final Event event, final EventCollector eventCollector) throws Exception {
- LOGGER.info("DoNothingProcessor process event: {}", event);
+ public void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector)
+ throws IOException {
+ eventCollector.collect(tsFileInsertionEvent);
+ }
+
+ @Override
+ public void process(Event event, EventCollector eventCollector) throws IOException {
eventCollector.collect(event);
}
@Override
- public void close() throws Exception {
- // Do Nothing
+ public void close() {
+ // do nothing
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/sink/SessionSink.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/sink/DemoSink.java
similarity index 56%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/sink/SessionSink.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/sink/DemoSink.java
index 64966db..8373b86 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/sink/SessionSink.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/sink/DemoSink.java
@@ -17,9 +17,8 @@
* under the License.
*/
-package org.apache.iotdb.collector.plugin.sink;
+package org.apache.iotdb.collector.plugin.builtin.sink;
-import org.apache.iotdb.collector.plugin.event.SourceEvent;
import org.apache.iotdb.pipe.api.PipeSink;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeSinkRuntimeConfiguration;
@@ -27,44 +26,60 @@ import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class SessionSink implements PipeSink {
+public class DemoSink implements PipeSink {
- private static final Logger LOGGER = LoggerFactory.getLogger(SessionSink.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(DemoSink.class);
@Override
- public void validate(final PipeParameterValidator validator) throws Exception {}
+ public void validate(final PipeParameterValidator validator) {
+ LOGGER.info("DemoSink validate successfully");
+ }
+ // TODO: SHIT..
@Override
- public void customize(
- final PipeParameters parameters, final PipeConnectorRuntimeConfiguration configuration)
- throws Exception {}
+ public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration)
+ throws Exception {
+ LOGGER.info("DemoSink customize successfully");
+ }
@Override
public void customize(
- final PipeParameters parameters, final PipeSinkRuntimeConfiguration configuration)
- throws Exception {}
+ final PipeParameters parameters, final PipeSinkRuntimeConfiguration configuration) {
+ LOGGER.info("DemoSink customize successfully");
+ }
+
+ @Override
+ public void handshake() {
+ LOGGER.info("DemoSink handshake successfully");
+ }
@Override
- public void handshake() throws Exception {
- LOGGER.info("SessionSink handshake successfully");
+ public void heartbeat() {
+ LOGGER.info("DemoSink heartbeat successfully");
}
@Override
- public void heartbeat() throws Exception {}
+ public void transfer(final TabletInsertionEvent tabletInsertionEvent) {
+ LOGGER.info("DemoSink transfer TabletInsertionEvent successfully");
+ }
@Override
- public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exception {}
+ public void transfer(TsFileInsertionEvent tsFileInsertionEvent) {
+ LOGGER.info("DemoSink transfer TsFileInsertionEvent successfully");
+ }
@Override
- public void transfer(final Event event) throws Exception {
- final SourceEvent sourceEvent = (SourceEvent) event;
- LOGGER.info("SessionSink transfer successfully {}", sourceEvent);
+ public void transfer(final Event event) {
+ LOGGER.info("DemoSink transfer successfully {}", event);
}
@Override
- public void close() throws Exception {}
+ public void close() throws Exception {
+ LOGGER.info("DemoSink close successfully");
+ }
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpPullSource.java
similarity index 86%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpPullSource.java
index c809b5f..c8bfd93 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpPullSource.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.collector.plugin.source;
+package org.apache.iotdb.collector.plugin.builtin.source;
import org.apache.iotdb.collector.plugin.api.PullSource;
-import org.apache.iotdb.collector.plugin.event.SourceEvent;
+import org.apache.iotdb.collector.plugin.api.event.DemoEvent;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -56,9 +56,9 @@ public class HttpPullSource extends PullSource {
@Override
public Event supply() {
- final Event event = new SourceEvent(String.valueOf(new Random().nextInt(1000)));
- LOGGER.info("event: {} created success", event);
- LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2));
+ LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(100));
+ final Event event = new DemoEvent(String.valueOf(new Random().nextInt(1000)));
+ LOGGER.info("{} created successfully ...", event);
return event;
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpPushSource.java
similarity index 90%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpPushSource.java
index 7ad2cc3..233335c 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpPushSource.java
@@ -17,10 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.collector.plugin.source;
+package org.apache.iotdb.collector.plugin.builtin.source;
import org.apache.iotdb.collector.plugin.api.PushSource;
-import org.apache.iotdb.collector.plugin.event.SourceEvent;
+import org.apache.iotdb.collector.plugin.api.event.DemoEvent;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -63,10 +63,10 @@ public class HttpPushSource extends PushSource {
private void doWork() {
try {
while (isStarted && !Thread.currentThread().isInterrupted()) {
- final Event event = new SourceEvent(String.valueOf(new Random().nextInt(1000)));
- LOGGER.info("event: {} created success", event);
+ final Event event = new DemoEvent(String.valueOf(new Random().nextInt(1000)));
+ LOGGER.info("{} created successfully ...", event);
supply(event);
- TimeUnit.SECONDS.sleep(2);
+ TimeUnit.SECONDS.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java
index 7c3bdc0..43620b3 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/ProcessorConstructor.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.collector.runtime.plugin.constructor;
-import org.apache.iotdb.collector.plugin.BuiltinPlugin;
-import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor;
+import org.apache.iotdb.collector.plugin.builtin.BuiltinPlugin;
+import org.apache.iotdb.collector.plugin.builtin.processor.DoNothingProcessor;
import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java
index dc640ac..066cf08 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SinkConstructor.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.collector.runtime.plugin.constructor;
-import org.apache.iotdb.collector.plugin.BuiltinPlugin;
-import org.apache.iotdb.collector.plugin.sink.SessionSink;
+import org.apache.iotdb.collector.plugin.builtin.BuiltinPlugin;
+import org.apache.iotdb.collector.plugin.builtin.sink.DemoSink;
import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
import org.apache.iotdb.pipe.api.PipeSink;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
@@ -33,7 +33,7 @@ public class SinkConstructor extends PluginConstructor {
@Override
protected void initConstructors() {
- pluginConstructors.put(BuiltinPlugin.IOTDB_THRIFT_SINK.getPluginName(), SessionSink::new);
+ pluginConstructors.put(BuiltinPlugin.IOTDB_THRIFT_SINK.getPluginName(), DemoSink::new);
}
@Override
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java
index 871dc00..f4c2cd0 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/constructor/SourceConstructor.java
@@ -19,11 +19,11 @@
package org.apache.iotdb.collector.runtime.plugin.constructor;
-import org.apache.iotdb.collector.plugin.BuiltinPlugin;
import org.apache.iotdb.collector.plugin.api.PullSource;
import org.apache.iotdb.collector.plugin.api.PushSource;
-import org.apache.iotdb.collector.plugin.source.HttpPullSource;
-import org.apache.iotdb.collector.plugin.source.HttpPushSource;
+import org.apache.iotdb.collector.plugin.builtin.BuiltinPlugin;
+import org.apache.iotdb.collector.plugin.builtin.source.HttpPullSource;
+import org.apache.iotdb.collector.plugin.builtin.source.HttpPushSource;
import org.apache.iotdb.collector.runtime.plugin.meta.PluginMetaKeeper;
import org.apache.iotdb.pipe.api.PipeSource;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkExceptionHandler.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkExceptionHandler.java
index bcaea48..d567337 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkExceptionHandler.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkExceptionHandler.java
@@ -31,6 +31,7 @@ class SinkExceptionHandler implements ExceptionHandler<EventContainer> {
@Override
public void handleEventException(Throwable ex, long sequence, EventContainer event) {
+ // TODO: heartbeat
// TODO: retry strategy
LOGGER.warn("Failed to handle event", ex);
}
(iotdb-extras) 01/03: refactor
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch polish-collector
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
commit cc61e0636f2b03f7597f53985d93e1f838e5803a
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Mar 5 01:45:42 2025 +0800
refactor
---
.../CollectorProcessorRuntimeConfiguration.java} | 29 +++--
.../customizer/CollectorRuntimeEnvironment.java | 59 +++++++++
.../CollectorSinkRuntimeConfiguration.java} | 29 +++--
.../CollectorSourceRuntimeConfiguration.java} | 29 +++--
.../collector/runtime/task/{def => }/Task.java | 41 +++++--
.../runtime/task/{def => }/TaskCombiner.java | 2 +-
.../iotdb/collector/runtime/task/TaskRuntime.java | 15 ++-
.../runtime/task/def/processor/ProcessorTask.java | 119 -------------------
.../collector/runtime/task/def/sink/SinkTask.java | 104 ----------------
.../runtime/task/def/source/PullSourceTask.java | 82 -------------
.../runtime/task/def/source/SourceTask.java | 132 ---------------------
.../task/{execution => event}/EventCollector.java | 7 +-
.../task/{execution => event}/EventContainer.java | 2 +-
.../runtime/task/execution/EventConsumer.java | 78 ------------
.../runtime/task/processor/ProcessorConsumer.java | 57 +++++++++
.../ProcessorExceptionHandler.java} | 18 +--
.../runtime/task/processor/ProcessorTask.java | 119 +++++++++++++++++++
.../collector/runtime/task/sink/SinkConsumer.java | 54 +++++++++
.../SinkExceptionHandler.java} | 18 +--
.../collector/runtime/task/sink/SinkTask.java | 106 +++++++++++++++++
.../EventCollector.java => source/SourceTask.java} | 33 +++---
.../pull/PullSourceConsumer.java} | 24 +++-
.../runtime/task/source/pull/PullSourceTask.java | 121 +++++++++++++++++++
.../source => source/push}/PushSourceTask.java | 13 +-
24 files changed, 663 insertions(+), 628 deletions(-)
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorProcessorRuntimeConfiguration.java
similarity index 51%
copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorProcessorRuntimeConfiguration.java
index 3fee53a..c135801 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorProcessorRuntimeConfiguration.java
@@ -17,27 +17,26 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.plugin.api.customizer;
-import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment;
-import com.lmax.disruptor.RingBuffer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+public class CollectorProcessorRuntimeConfiguration implements PipeProcessorRuntimeConfiguration {
-public class EventCollector implements org.apache.iotdb.pipe.api.collector.EventCollector {
+ private final CollectorRuntimeEnvironment runtimeEnvironment;
- private static final Logger LOGGER = LoggerFactory.getLogger(EventCollector.class);
-
- private final RingBuffer<EventContainer> ringBuffer;
-
- public EventCollector(final RingBuffer<EventContainer> ringBuffer) {
- this.ringBuffer = ringBuffer;
+ public CollectorProcessorRuntimeConfiguration(
+ final String pipeName,
+ final long creationTime,
+ final int parallelism,
+ final int instanceIndex) {
+ runtimeEnvironment =
+ new CollectorRuntimeEnvironment(pipeName, creationTime, parallelism, instanceIndex);
}
@Override
- public void collect(final Event event) {
- ringBuffer.publishEvent((container, sequence, o) -> container.setEvent(event), event);
- LOGGER.info("successfully publish event {}", event);
+ public PipeRuntimeEnvironment getRuntimeEnvironment() {
+ return runtimeEnvironment;
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorRuntimeEnvironment.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorRuntimeEnvironment.java
new file mode 100644
index 0000000..8ace81c
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorRuntimeEnvironment.java
@@ -0,0 +1,59 @@
+/*
+ * 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.collector.plugin.api.customizer;
+
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment;
+
+public class CollectorRuntimeEnvironment implements PipeRuntimeEnvironment {
+
+ private final String pipeName;
+ private final long creationTime;
+ private final int parallelism;
+ private final int instanceIndex;
+
+ public CollectorRuntimeEnvironment(
+ final String pipeName,
+ final long creationTime,
+ final int parallelism,
+ final int instanceIndex) {
+ this.pipeName = pipeName;
+ this.creationTime = creationTime;
+ this.parallelism = parallelism;
+ this.instanceIndex = instanceIndex;
+ }
+
+ @Override
+ public String getPipeName() {
+ return pipeName;
+ }
+
+ @Override
+ public long getCreationTime() {
+ return creationTime;
+ }
+
+ public int getParallelism() {
+ return parallelism;
+ }
+
+ public int getInstanceIndex() {
+ return instanceIndex;
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorSinkRuntimeConfiguration.java
similarity index 52%
copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorSinkRuntimeConfiguration.java
index 3fee53a..b709291 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorSinkRuntimeConfiguration.java
@@ -17,27 +17,26 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.plugin.api.customizer;
-import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeSinkRuntimeConfiguration;
-import com.lmax.disruptor.RingBuffer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+public class CollectorSinkRuntimeConfiguration implements PipeSinkRuntimeConfiguration {
-public class EventCollector implements org.apache.iotdb.pipe.api.collector.EventCollector {
+ private final CollectorRuntimeEnvironment runtimeEnvironment;
- private static final Logger LOGGER = LoggerFactory.getLogger(EventCollector.class);
-
- private final RingBuffer<EventContainer> ringBuffer;
-
- public EventCollector(final RingBuffer<EventContainer> ringBuffer) {
- this.ringBuffer = ringBuffer;
+ public CollectorSinkRuntimeConfiguration(
+ final String pipeName,
+ final long creationTime,
+ final int parallelism,
+ final int instanceIndex) {
+ runtimeEnvironment =
+ new CollectorRuntimeEnvironment(pipeName, creationTime, parallelism, instanceIndex);
}
@Override
- public void collect(final Event event) {
- ringBuffer.publishEvent((container, sequence, o) -> container.setEvent(event), event);
- LOGGER.info("successfully publish event {}", event);
+ public PipeRuntimeEnvironment getRuntimeEnvironment() {
+ return runtimeEnvironment;
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorSourceRuntimeConfiguration.java
similarity index 52%
copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorSourceRuntimeConfiguration.java
index 3fee53a..93970f9 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/customizer/CollectorSourceRuntimeConfiguration.java
@@ -17,27 +17,26 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.plugin.api.customizer;
-import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment;
+import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration;
-import com.lmax.disruptor.RingBuffer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+public class CollectorSourceRuntimeConfiguration implements PipeSourceRuntimeConfiguration {
-public class EventCollector implements org.apache.iotdb.pipe.api.collector.EventCollector {
+ private final CollectorRuntimeEnvironment runtimeEnvironment;
- private static final Logger LOGGER = LoggerFactory.getLogger(EventCollector.class);
-
- private final RingBuffer<EventContainer> ringBuffer;
-
- public EventCollector(final RingBuffer<EventContainer> ringBuffer) {
- this.ringBuffer = ringBuffer;
+ public CollectorSourceRuntimeConfiguration(
+ final String pipeName,
+ final long creationTime,
+ final int parallelism,
+ final int instanceIndex) {
+ runtimeEnvironment =
+ new CollectorRuntimeEnvironment(pipeName, creationTime, parallelism, instanceIndex);
}
@Override
- public void collect(final Event event) {
- ringBuffer.publishEvent((container, sequence, o) -> container.setEvent(event), event);
- LOGGER.info("successfully publish event {}", event);
+ public PipeRuntimeEnvironment getRuntimeEnvironment() {
+ return runtimeEnvironment;
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/Task.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/Task.java
similarity index 65%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/Task.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/Task.java
index 960a6ef..610e8ae 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/Task.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/Task.java
@@ -17,26 +17,40 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.def;
+package org.apache.iotdb.collector.runtime.task;
-import org.apache.iotdb.collector.runtime.task.execution.EventCollector;
-import org.apache.iotdb.collector.runtime.task.execution.EventConsumer;
-import org.apache.iotdb.pipe.api.PipePlugin;
+import org.apache.iotdb.collector.config.TaskRuntimeOptions;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Objects;
+import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;
-import java.util.stream.Stream;
public abstract class Task {
private static final Logger LOGGER = LoggerFactory.getLogger(Task.class);
+ protected final String taskId;
+ protected final PipeParameters parameters;
+
+ protected final int parallelism;
+
private static final long CHECK_RUNNING_INTERVAL_NANOS = 100_000_000L;
- private final AtomicBoolean isRunning = new AtomicBoolean(false);
+ protected final AtomicBoolean isRunning = new AtomicBoolean(false);
+ protected final AtomicBoolean isDropped = new AtomicBoolean(false);
+
+ protected Task(final String taskId, final Map<String, String> attributes) {
+ this.taskId = taskId;
+ this.parameters = new PipeParameters(attributes);
+
+ this.parallelism =
+ parameters.getIntOrDefault(
+ TaskRuntimeOptions.TASK_SINK_PARALLELISM_NUM.key(),
+ TaskRuntimeOptions.TASK_SINK_PARALLELISM_NUM.value());
+ }
public void resume() {
isRunning.set(true);
@@ -46,13 +60,13 @@ public abstract class Task {
isRunning.set(false);
}
- public void waitUntilRunning() {
- while (!isRunning.get()) {
+ protected void waitUntilRunningOrDropped() {
+ while (!isRunning.get() && !isDropped.get()) {
LockSupport.parkNanos(CHECK_RUNNING_INTERVAL_NANOS);
}
}
- public final void create() {
+ public final synchronized void create() {
try {
resume();
createInternal();
@@ -63,7 +77,7 @@ public abstract class Task {
public abstract void createInternal() throws Exception;
- public final void start() {
+ public final synchronized void start() {
try {
resume();
startInternal();
@@ -74,7 +88,7 @@ public abstract class Task {
public abstract void startInternal() throws Exception;
- public final void stop() {
+ public final synchronized void stop() {
try {
pause();
stopInternal();
@@ -85,9 +99,10 @@ public abstract class Task {
public abstract void stopInternal() throws Exception;
- public final void drop() {
+ public final synchronized void drop() {
try {
pause();
+ isDropped.set(true);
dropInternal();
} catch (final Exception e) {
LOGGER.warn("Failed to drop task", e);
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskCombiner.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskCombiner.java
similarity index 97%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskCombiner.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskCombiner.java
index e6b6a43..b100160 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskCombiner.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskCombiner.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.def;
+package org.apache.iotdb.collector.runtime.task;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java
index dd09a1c..178a3ae 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java
@@ -19,11 +19,9 @@
package org.apache.iotdb.collector.runtime.task;
-import org.apache.iotdb.collector.runtime.task.def.TaskCombiner;
-import org.apache.iotdb.collector.runtime.task.def.processor.ProcessorTask;
-import org.apache.iotdb.collector.runtime.task.def.sink.SinkTask;
-import org.apache.iotdb.collector.runtime.task.def.source.PushSourceTask;
-import org.apache.iotdb.collector.runtime.task.def.source.SourceTask;
+import org.apache.iotdb.collector.runtime.task.processor.ProcessorTask;
+import org.apache.iotdb.collector.runtime.task.sink.SinkTask;
+import org.apache.iotdb.collector.runtime.task.source.SourceTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,9 +49,10 @@ public class TaskRuntime implements AutoCloseable {
.build();
}
- final SinkTask sinkTask = new SinkTask(sinkAttribute);
- final ProcessorTask processorTask = new ProcessorTask(processorAttribute, sinkTask);
- final SourceTask sourceTask = new PushSourceTask(taskId, sourceAttribute, processorTask);
+ final SinkTask sinkTask = new SinkTask(taskId, sinkAttribute);
+ final ProcessorTask processorTask =
+ new ProcessorTask(taskId, processorAttribute, sinkTask.makeProducer());
+ final SourceTask sourceTask = SourceTask.construct(taskId, sourceAttribute, processorTask);
final TaskCombiner taskRepository = new TaskCombiner(sourceTask, processorTask, sinkTask);
tasks.put(taskId, taskRepository);
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/processor/ProcessorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/processor/ProcessorTask.java
deleted file mode 100644
index 3e80847..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/processor/ProcessorTask.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * 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.collector.runtime.task.def.processor;
-
-import org.apache.iotdb.collector.config.TaskRuntimeOptions;
-import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor;
-import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
-import org.apache.iotdb.collector.runtime.task.def.Task;
-import org.apache.iotdb.collector.runtime.task.def.sink.SinkTask;
-import org.apache.iotdb.collector.runtime.task.execution.EventConsumerExceptionHandler;
-import org.apache.iotdb.collector.runtime.task.execution.EventCollector;
-import org.apache.iotdb.collector.runtime.task.execution.EventConsumer;
-import org.apache.iotdb.collector.runtime.task.execution.EventContainer;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-
-import com.lmax.disruptor.BlockingWaitStrategy;
-import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.dsl.Disruptor;
-import com.lmax.disruptor.dsl.ProducerType;
-import com.lmax.disruptor.util.DaemonThreadFactory;
-
-import java.util.Map;
-import java.util.Optional;
-
-public class ProcessorTask extends Task {
-
- private Disruptor<EventContainer> processorDisruptor;
- private EventConsumer[] eventConsumers;
-
- private final PipeParameters parameters;
- private final int processParallelismNum;
- private final EventCollector collector;
-
- public ProcessorTask(final Map<String, String> processorAttributes, final SinkTask sinkTask) {
- this.parameters = new PipeParameters(processorAttributes);
- this.processParallelismNum =
- this.parameters.getIntOrDefault(
- TaskRuntimeOptions.TASK_PROCESS_PARALLELISM_NUM.key(),
- TaskRuntimeOptions.TASK_PROCESS_PARALLELISM_NUM.value());
- this.collector = new EventCollector(sinkTask.getSinkRingBuffer());
-
- this.initProcessorDisruptor();
- }
-
- @Override
- public void createInternal() {
- if (this.processorDisruptor == null) {
- this.initProcessorDisruptor();
- }
-
- this.eventConsumers =
- this.getConsumer(
- PluginFactory.createInstance(DoNothingProcessor.class),
- processParallelismNum,
- collector);
-
- this.processorDisruptor.setDefaultExceptionHandler(new EventConsumerExceptionHandler());
- this.processorDisruptor.handleEventsWithWorkerPool(this.eventConsumers);
- this.processorDisruptor.start();
- }
-
- private void initProcessorDisruptor() {
- this.processorDisruptor =
- new Disruptor<>(
- EventContainer::new,
- this.parameters.getIntOrDefault(
- TaskRuntimeOptions.TASK_PROCESSOR_RING_BUFFER_SIZE.key(),
- TaskRuntimeOptions.TASK_PROCESSOR_RING_BUFFER_SIZE.value()),
- DaemonThreadFactory.INSTANCE,
- ProducerType.MULTI,
- new BlockingWaitStrategy());
- }
-
- @Override
- public void startInternal() throws Exception {
- for (final EventConsumer consumer : this.eventConsumers) {
- consumer.resume();
- }
- }
-
- @Override
- public void stopInternal() {
- for (final EventConsumer consumer : this.eventConsumers) {
- consumer.pause();
- }
- }
-
- @Override
- public void dropInternal() {
- if (this.processorDisruptor != null) {
- this.processorDisruptor.shutdown();
- this.processorDisruptor = null;
- }
- }
-
- public Optional<RingBuffer<EventContainer>> getProcessorRingBuffer() {
- if (this.processorDisruptor != null) {
- return Optional.of(this.processorDisruptor.getRingBuffer());
- }
- return Optional.empty();
- }
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkTask.java
deleted file mode 100644
index f83bc31..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkTask.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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.collector.runtime.task.def.sink;
-
-import com.lmax.disruptor.WorkHandler;
-import org.apache.iotdb.collector.config.TaskRuntimeOptions;
-import org.apache.iotdb.collector.plugin.sink.SessionSink;
-import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
-import org.apache.iotdb.collector.runtime.task.def.Task;
-import org.apache.iotdb.collector.runtime.task.execution.EventCollector;
-import org.apache.iotdb.collector.runtime.task.execution.EventConsumerExceptionHandler;
-import org.apache.iotdb.collector.runtime.task.execution.EventConsumer;
-import org.apache.iotdb.collector.runtime.task.execution.EventContainer;
-import org.apache.iotdb.pipe.api.PipePlugin;
-import org.apache.iotdb.pipe.api.PipeSink;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-
-import com.lmax.disruptor.BlockingWaitStrategy;
-import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.dsl.Disruptor;
-import com.lmax.disruptor.dsl.ProducerType;
-import com.lmax.disruptor.util.DaemonThreadFactory;
-
-import java.util.Map;
-import java.util.Objects;
-import java.util.stream.Stream;
-
-public class SinkTask extends Task {
-
- private static class SinkConsumer implements WorkHandler<EventContainer> {
-
- private final PipeSink sink;
-
- private SinkConsumer(final PipeSink sink) {
- this.sink = sink;
- }
-
- @Override
- public void onEvent(EventContainer event) throws Exception {
- sink.transfer(event.getEvent());
- }
- }
-
- private final Disruptor<EventContainer> disruptor;
- private final int parallelism;
-
- public SinkTask(final Map<String, String> processorAttributes) {
- final PipeParameters parameters = new PipeParameters(processorAttributes);
- parallelism =
- parameters.getIntOrDefault(
- TaskRuntimeOptions.TASK_SINK_PARALLELISM_NUM.key(),
- TaskRuntimeOptions.TASK_SINK_PARALLELISM_NUM.value());
- disruptor = new Disruptor<>(
- EventContainer::new,
- parallelism,
- DaemonThreadFactory.INSTANCE,
- ProducerType.MULTI,
- new BlockingWaitStrategy());
- }
-
- @Override
- public void createInternal() {
- disruptor.setDefaultExceptionHandler(new EventConsumerExceptionHandler());
- disruptor.handleEventsWithWorkerPool(Stream.generate(() -> new SinkConsumer(PluginFactory.createInstance(SessionSink.class)))
- .limit(parallelism)
- .toArray(SinkConsumer[]::new));
- }
-
- @Override
- public void startInternal() {
- disruptor.start();
- }
-
- @Override
- public void stopInternal() {
- disruptor.halt();
- }
-
- @Override
- public void dropInternal() {
- disruptor.shutdown();
- }
-
- public RingBuffer<EventContainer> getSinkRingBuffer() {
- return this.disruptor.getRingBuffer();
- }
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PullSourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PullSourceTask.java
deleted file mode 100644
index 31e919e..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PullSourceTask.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * 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.collector.runtime.task.def.source;
-
-import org.apache.iotdb.collector.plugin.source.HttpPullSource;
-import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
-import org.apache.iotdb.collector.runtime.task.def.processor.ProcessorTask;
-import org.apache.iotdb.pipe.api.event.Event;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.LockSupport;
-
-public class PullSourceTask extends SourceTask {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(PullSourceTask.class);
-
- public PullSourceTask(
- final String taskId,
- final Map<String, String> sourceParams,
- final ProcessorTask processorTask) {
- super(taskId, sourceParams, processorTask);
- }
-
- @Override
- public void createInternal() {
- createSourceTask();
- for (int i = 0; i < sourceParallelismNum; i++) {
- // use sourceAttribute later
- try (final HttpPullSource source = PluginFactory.createInstance(HttpPullSource.class)) {
- addSourceTask(source);
- SOURCE_EXECUTOR_SERVICE
- .get(taskId)
- .submit(
- () -> {
- try {
- source.start();
- while (SOURCE_TASK_STATUS.containsKey(taskId)) {
- if (SOURCE_TASK_STATUS.get(taskId) == TaskState.Stopped) {
- LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
- continue;
- }
-
- final Event event = source.supply();
- processorTask
- .getProcessorRingBuffer()
- .ifPresent(
- ringBuffer ->
- ringBuffer.publishEvent(
- ((container, sequence, o) -> container.setEvent(event)),
- event));
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- } catch (final Exception e) {
- LOGGER.warn("Pull source task failed", e);
- }
- }
- }
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/SourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/SourceTask.java
deleted file mode 100644
index 921405b..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/SourceTask.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * 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.collector.runtime.task.def.source;
-
-import org.apache.iotdb.collector.config.TaskRuntimeOptions;
-import org.apache.iotdb.collector.runtime.task.def.Task;
-import org.apache.iotdb.collector.runtime.task.def.processor.ProcessorTask;
-import org.apache.iotdb.commons.concurrent.IoTThreadFactory;
-import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor;
-import org.apache.iotdb.pipe.api.PipeSource;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-public abstract class SourceTask extends Task {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(SourceTask.class);
-
- // Source needs the processor's RingBuffer to publish event.
- protected final ProcessorTask processorTask;
- protected final int sourceParallelismNum;
- protected final String taskId;
- // Store the status of the tasks, Running or Stopped
- protected static final Map<String, TaskState> SOURCE_TASK_STATUS = new ConcurrentHashMap<>();
- protected static final Map<String, ExecutorService> SOURCE_EXECUTOR_SERVICE =
- new ConcurrentHashMap<>();
- // Source tasks list
- protected static final Map<String, List<PipeSource>> SOURCE_TASK = new ConcurrentHashMap<>();
-
- public SourceTask(
- final String taskId,
- final Map<String, String> sourceParams,
- final ProcessorTask processorTask) {
- this.taskId = taskId;
- final PipeParameters params = new PipeParameters(sourceParams);
- this.sourceParallelismNum =
- params.getIntOrDefault(
- TaskRuntimeOptions.TASK_SOURCE_PARALLELISM_NUM.key(),
- TaskRuntimeOptions.TASK_SOURCE_PARALLELISM_NUM.value());
- this.processorTask = processorTask;
- }
-
- @Override
- public void startInternal() throws Exception {
- SOURCE_TASK_STATUS.computeIfPresent(taskId, (taskId, status) -> TaskState.Running);
- SOURCE_TASK
- .get(taskId)
- .forEach(
- source -> {
- try {
- source.start();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- }
-
- @Override
- public synchronized void stopInternal() {
- SOURCE_TASK_STATUS.computeIfPresent(taskId, (taskId, status) -> TaskState.Stopped);
- SOURCE_TASK
- .get(taskId)
- .forEach(
- source -> {
- try {
- source.close();
- } catch (Exception e) {
- LOGGER.warn("Failed to close source", e);
- }
- });
- }
-
- @Override
- public void dropInternal() {
- stopInternal();
- SOURCE_TASK.remove(taskId);
- SOURCE_TASK_STATUS.remove(taskId);
- SOURCE_EXECUTOR_SERVICE.remove(taskId).shutdown();
- }
-
- protected void addSourceTask(final PipeSource source) {
- if (!SOURCE_TASK.containsKey(taskId)) {
- SOURCE_TASK.put(taskId, new CopyOnWriteArrayList<>());
- }
- SOURCE_TASK.get(taskId).add(source);
- }
-
- protected void createSourceTask() {
- SOURCE_EXECUTOR_SERVICE.putIfAbsent(
- taskId,
- new WrappedThreadPoolExecutor(
- sourceParallelismNum,
- sourceParallelismNum,
- 0L,
- TimeUnit.SECONDS,
- new LinkedBlockingQueue<>(),
- new IoTThreadFactory("source-executor"),
- "source-executor"));
- SOURCE_TASK_STATUS.putIfAbsent(taskId, TaskState.Running);
- }
-
- protected enum TaskState {
- Running,
- Stopped
- }
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventCollector.java
similarity index 83%
copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventCollector.java
index 3fee53a..2e2675a 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventCollector.java
@@ -17,18 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.runtime.task.event;
import org.apache.iotdb.pipe.api.event.Event;
import com.lmax.disruptor.RingBuffer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class EventCollector implements org.apache.iotdb.pipe.api.collector.EventCollector {
- private static final Logger LOGGER = LoggerFactory.getLogger(EventCollector.class);
-
private final RingBuffer<EventContainer> ringBuffer;
public EventCollector(final RingBuffer<EventContainer> ringBuffer) {
@@ -38,6 +34,5 @@ public class EventCollector implements org.apache.iotdb.pipe.api.collector.Event
@Override
public void collect(final Event event) {
ringBuffer.publishEvent((container, sequence, o) -> container.setEvent(event), event);
- LOGGER.info("successfully publish event {}", event);
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventContainer.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventContainer.java
similarity index 94%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventContainer.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventContainer.java
index 60beb40..99817e7 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventContainer.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/event/EventContainer.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.runtime.task.event;
import org.apache.iotdb.pipe.api.event.Event;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumer.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumer.java
deleted file mode 100644
index be10c92..0000000
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumer.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.collector.runtime.task.execution;
-
-import org.apache.iotdb.pipe.api.PipePlugin;
-import org.apache.iotdb.pipe.api.PipeProcessor;
-import org.apache.iotdb.pipe.api.PipeSink;
-
-import com.lmax.disruptor.WorkHandler;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.LockSupport;
-
-public class EventConsumer implements WorkHandler<EventContainer> {
-
- private final PipePlugin plugin;
- private final EventCollector collector;
-
- private final AtomicBoolean running = new AtomicBoolean(true);
-
- private static final long PARK_NANOS = 100_000_000L;
-
- public void pause() {
- running.set(false);
- }
-
- public void resume() {
- running.set(true);
- }
-
- public boolean shouldRun() {
- while (!running.get()) {
- LockSupport.parkNanos(PARK_NANOS);
- }
- return running.get();
- }
-
- public EventConsumer(
- final PipePlugin plugin,
- final EventCollector collector) {
- this.plugin = plugin;
- this.collector = collector;
- }
-
- public EventConsumer(
- final PipePlugin plugin) {
- this(plugin, null);
- }
-
- @Override
- public void onEvent(final EventContainer eventContainer) throws Exception {
- if (!shouldRun()) {
- return;
- }
- if (plugin instanceof PipeProcessor) {
- ((PipeProcessor) plugin).process(eventContainer.getEvent(), this.collector);
- } else if (plugin instanceof PipeSink) {
- ((PipeSink) plugin).transfer(eventContainer.getEvent());
- }
- }
-}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorConsumer.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorConsumer.java
new file mode 100644
index 0000000..c153506
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorConsumer.java
@@ -0,0 +1,57 @@
+/*
+ * 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.collector.runtime.task.processor;
+
+import org.apache.iotdb.collector.runtime.task.event.EventContainer;
+import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.collector.EventCollector;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
+
+import com.lmax.disruptor.WorkHandler;
+
+class ProcessorConsumer implements WorkHandler<EventContainer> {
+
+ private final PipeProcessor processor;
+ private final EventCollector eventCollector;
+
+ ProcessorConsumer(final PipeProcessor processor, final EventCollector eventCollector) {
+ this.processor = processor;
+ this.eventCollector = eventCollector;
+ }
+
+ PipeProcessor consumer() {
+ return processor;
+ }
+
+ @Override
+ public void onEvent(final EventContainer eventContainer) throws Exception {
+ // TODO: retry strategy
+ final Event event = eventContainer.getEvent();
+ if (event instanceof TabletInsertionEvent) {
+ processor.process((TabletInsertionEvent) event, eventCollector);
+ } else if (event instanceof TsFileInsertionEvent) {
+ processor.process((TsFileInsertionEvent) event, eventCollector);
+ } else if (event != null) {
+ processor.process(event, eventCollector);
+ }
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumerExceptionHandler.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorExceptionHandler.java
similarity index 68%
copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumerExceptionHandler.java
copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorExceptionHandler.java
index 922383e..277cdb0 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumerExceptionHandler.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorExceptionHandler.java
@@ -17,27 +17,31 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.runtime.task.processor;
+
+import org.apache.iotdb.collector.runtime.task.event.EventContainer;
import com.lmax.disruptor.ExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class EventConsumerExceptionHandler implements ExceptionHandler<Object> {
- private static final Logger LOGGER = LoggerFactory.getLogger(EventConsumerExceptionHandler.class);
+class ProcessorExceptionHandler implements ExceptionHandler<EventContainer> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ProcessorExceptionHandler.class);
@Override
- public void handleEventException(Throwable ex, long sequence, Object event) {
- LOGGER.error("Event processing failed [seq={}, event={}]", sequence, event, ex);
+ public void handleEventException(Throwable ex, long sequence, EventContainer event) {
+ // TODO: retry strategy
+ LOGGER.warn("Failed to handle event", ex);
}
@Override
public void handleOnStartException(Throwable ex) {
- LOGGER.error("Failed to start disruptor", ex);
+ LOGGER.warn("Failed to start processor disruptor", ex);
}
@Override
public void handleOnShutdownException(Throwable ex) {
- LOGGER.error("Failed to shutdown disruptor", ex);
+ LOGGER.warn("Failed to shutdown processor disruptor", ex);
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java
new file mode 100644
index 0000000..51def10
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/processor/ProcessorTask.java
@@ -0,0 +1,119 @@
+/*
+ * 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.collector.runtime.task.processor;
+
+import org.apache.iotdb.collector.plugin.api.customizer.CollectorProcessorRuntimeConfiguration;
+import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor;
+import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
+import org.apache.iotdb.collector.runtime.task.Task;
+import org.apache.iotdb.collector.runtime.task.event.EventCollector;
+import org.apache.iotdb.collector.runtime.task.event.EventContainer;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import com.lmax.disruptor.BlockingWaitStrategy;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.dsl.ProducerType;
+import com.lmax.disruptor.util.DaemonThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+public class ProcessorTask extends Task {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ProcessorTask.class);
+
+ private final Disruptor<EventContainer> disruptor;
+ private final EventCollector sinkProducer;
+ private ProcessorConsumer[] processorConsumers;
+
+ public ProcessorTask(
+ final String taskId,
+ final Map<String, String> attributes,
+ final EventCollector sinkProducer) {
+ super(taskId, attributes);
+
+ disruptor =
+ new Disruptor<>(
+ EventContainer::new,
+ parallelism, // fault usage
+ DaemonThreadFactory.INSTANCE, // fault usage
+ ProducerType.MULTI,
+ new BlockingWaitStrategy());
+ this.sinkProducer = sinkProducer;
+ }
+
+ @Override
+ public void createInternal() throws Exception {
+ final long creationTime = System.currentTimeMillis();
+ processorConsumers = new ProcessorConsumer[parallelism];
+ for (int i = 0; i < parallelism; i++) {
+ // TODO: PluginFactory
+ processorConsumers[i] =
+ new ProcessorConsumer(
+ PluginFactory.createInstance(DoNothingProcessor.class), sinkProducer);
+ processorConsumers[i].consumer().validate(new PipeParameterValidator(parameters));
+ processorConsumers[i]
+ .consumer()
+ .customize(
+ parameters,
+ new CollectorProcessorRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ }
+ disruptor.handleEventsWithWorkerPool(processorConsumers);
+
+ disruptor.setDefaultExceptionHandler(new ProcessorExceptionHandler());
+ }
+
+ @Override
+ public void startInternal() {
+ disruptor.start();
+ }
+
+ @Override
+ public void stopInternal() {
+ disruptor.halt();
+ }
+
+ @Override
+ public void dropInternal() {
+ if (processorConsumers != null) {
+ for (int i = 0; i < parallelism; i++) {
+ try {
+ processorConsumers[i].consumer().close();
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to close sink", e);
+ }
+ }
+ }
+
+ disruptor.shutdown();
+ }
+
+ public EventCollector makeProducer() {
+ return new EventCollector(disruptor.getRingBuffer());
+ }
+
+ public void publishEvent(final Event event) {
+ disruptor
+ .getRingBuffer()
+ .publishEvent((container, sequence, o) -> container.setEvent(event), event);
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkConsumer.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkConsumer.java
new file mode 100644
index 0000000..2fe13f0
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkConsumer.java
@@ -0,0 +1,54 @@
+/*
+ * 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.collector.runtime.task.sink;
+
+import org.apache.iotdb.collector.runtime.task.event.EventContainer;
+import org.apache.iotdb.pipe.api.PipeSink;
+import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
+import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
+
+import com.lmax.disruptor.WorkHandler;
+
+class SinkConsumer implements WorkHandler<EventContainer> {
+
+ private final PipeSink sink;
+
+ SinkConsumer(final PipeSink sink) {
+ this.sink = sink;
+ }
+
+ PipeSink consumer() {
+ return sink;
+ }
+
+ @Override
+ public void onEvent(EventContainer eventContainer) throws Exception {
+ // TODO: retry strategy
+ final Event event = eventContainer.getEvent();
+ if (event instanceof TabletInsertionEvent) {
+ sink.transfer((TabletInsertionEvent) event);
+ } else if (event instanceof TsFileInsertionEvent) {
+ sink.transfer((TsFileInsertionEvent) event);
+ } else if (event != null) {
+ sink.transfer(event);
+ }
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumerExceptionHandler.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkExceptionHandler.java
similarity index 69%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumerExceptionHandler.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkExceptionHandler.java
index 922383e..bcaea48 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventConsumerExceptionHandler.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkExceptionHandler.java
@@ -17,27 +17,31 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.runtime.task.sink;
+
+import org.apache.iotdb.collector.runtime.task.event.EventContainer;
import com.lmax.disruptor.ExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class EventConsumerExceptionHandler implements ExceptionHandler<Object> {
- private static final Logger LOGGER = LoggerFactory.getLogger(EventConsumerExceptionHandler.class);
+class SinkExceptionHandler implements ExceptionHandler<EventContainer> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SinkExceptionHandler.class);
@Override
- public void handleEventException(Throwable ex, long sequence, Object event) {
- LOGGER.error("Event processing failed [seq={}, event={}]", sequence, event, ex);
+ public void handleEventException(Throwable ex, long sequence, EventContainer event) {
+ // TODO: retry strategy
+ LOGGER.warn("Failed to handle event", ex);
}
@Override
public void handleOnStartException(Throwable ex) {
- LOGGER.error("Failed to start disruptor", ex);
+ LOGGER.warn("Failed to start sink disruptor", ex);
}
@Override
public void handleOnShutdownException(Throwable ex) {
- LOGGER.error("Failed to shutdown disruptor", ex);
+ LOGGER.warn("Failed to shutdown sink disruptor", ex);
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java
new file mode 100644
index 0000000..fbda5a4
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/sink/SinkTask.java
@@ -0,0 +1,106 @@
+/*
+ * 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.collector.runtime.task.sink;
+
+import org.apache.iotdb.collector.plugin.api.customizer.CollectorSinkRuntimeConfiguration;
+import org.apache.iotdb.collector.plugin.sink.SessionSink;
+import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
+import org.apache.iotdb.collector.runtime.task.Task;
+import org.apache.iotdb.collector.runtime.task.event.EventCollector;
+import org.apache.iotdb.collector.runtime.task.event.EventContainer;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+
+import com.lmax.disruptor.BlockingWaitStrategy;
+import com.lmax.disruptor.dsl.Disruptor;
+import com.lmax.disruptor.dsl.ProducerType;
+import com.lmax.disruptor.util.DaemonThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+public class SinkTask extends Task {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SinkTask.class);
+
+ private final Disruptor<EventContainer> disruptor;
+ private SinkConsumer[] consumers;
+
+ public SinkTask(final String taskId, final Map<String, String> attributes) {
+ super(taskId, attributes);
+
+ disruptor =
+ new Disruptor<>(
+ EventContainer::new,
+ parallelism, // fault usage
+ DaemonThreadFactory.INSTANCE, // fault usage
+ ProducerType.MULTI,
+ new BlockingWaitStrategy());
+ }
+
+ @Override
+ public void createInternal() throws Exception {
+ final long creationTime = System.currentTimeMillis();
+ consumers = new SinkConsumer[parallelism];
+ for (int i = 0; i < parallelism; i++) {
+ // TODO: PluginFactory
+ consumers[i] = new SinkConsumer(PluginFactory.createInstance(SessionSink.class));
+ consumers[i].consumer().validate(new PipeParameterValidator(parameters));
+ consumers[i]
+ .consumer()
+ .customize(
+ parameters,
+ new CollectorSinkRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ consumers[i].consumer().handshake();
+ }
+ disruptor.handleEventsWithWorkerPool(consumers);
+
+ disruptor.setDefaultExceptionHandler(new SinkExceptionHandler());
+ }
+
+ @Override
+ public void startInternal() {
+ disruptor.start();
+ }
+
+ @Override
+ public void stopInternal() {
+ disruptor.halt();
+ }
+
+ @Override
+ public void dropInternal() {
+ if (consumers != null) {
+ for (int i = 0; i < parallelism; i++) {
+ try {
+ consumers[i].consumer().close();
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to close sink", e);
+ }
+ }
+ }
+
+ disruptor.shutdown();
+ }
+
+ public EventCollector makeProducer() {
+ return new EventCollector(disruptor.getRingBuffer());
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java
similarity index 50%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java
index 3fee53a..3b91cfa 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/execution/EventCollector.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/SourceTask.java
@@ -17,27 +17,30 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.execution;
+package org.apache.iotdb.collector.runtime.task.source;
-import org.apache.iotdb.pipe.api.event.Event;
+import org.apache.iotdb.collector.runtime.task.Task;
+import org.apache.iotdb.collector.runtime.task.processor.ProcessorTask;
+import org.apache.iotdb.collector.runtime.task.source.pull.PullSourceTask;
-import com.lmax.disruptor.RingBuffer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.Map;
-public class EventCollector implements org.apache.iotdb.pipe.api.collector.EventCollector {
+public abstract class SourceTask extends Task {
- private static final Logger LOGGER = LoggerFactory.getLogger(EventCollector.class);
+ protected final ProcessorTask processorTask;
- private final RingBuffer<EventContainer> ringBuffer;
-
- public EventCollector(final RingBuffer<EventContainer> ringBuffer) {
- this.ringBuffer = ringBuffer;
+ protected SourceTask(
+ final String taskId,
+ final Map<String, String> attributes,
+ final ProcessorTask processorTask) {
+ super(taskId, attributes);
+ this.processorTask = processorTask;
}
- @Override
- public void collect(final Event event) {
- ringBuffer.publishEvent((container, sequence, o) -> container.setEvent(event), event);
- LOGGER.info("successfully publish event {}", event);
+ public static SourceTask construct(
+ final String taskId,
+ final Map<String, String> attributes,
+ final ProcessorTask processorTask) {
+ return new PullSourceTask(taskId, attributes, processorTask);
}
}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkConsumer.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceConsumer.java
similarity index 54%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkConsumer.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceConsumer.java
index fc7d56f..d4f56f7 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/sink/SinkConsumer.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceConsumer.java
@@ -17,7 +17,27 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.def.sink;
+package org.apache.iotdb.collector.runtime.task.source.pull;
-public class SinkConsumer {
+import org.apache.iotdb.collector.plugin.api.CollectorPullSource;
+import org.apache.iotdb.pipe.api.collector.EventCollector;
+
+class PullSourceConsumer {
+
+ private final CollectorPullSource pullSource;
+ private final EventCollector eventCollector;
+
+ PullSourceConsumer(final CollectorPullSource pullSource, final EventCollector eventCollector) {
+ this.pullSource = pullSource;
+ this.eventCollector = eventCollector;
+ }
+
+ CollectorPullSource consumer() {
+ return pullSource;
+ }
+
+ public void onScheduler() throws Exception {
+ // TODO: scheduler strategy
+ eventCollector.collect(pullSource.supply());
+ }
}
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
new file mode 100644
index 0000000..bec9e75
--- /dev/null
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/pull/PullSourceTask.java
@@ -0,0 +1,121 @@
+/*
+ * 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.collector.runtime.task.source.pull;
+
+import org.apache.iotdb.collector.plugin.api.customizer.CollectorSourceRuntimeConfiguration;
+import org.apache.iotdb.collector.plugin.source.HttpPullSource;
+import org.apache.iotdb.collector.runtime.plugin.PluginFactory;
+import org.apache.iotdb.collector.runtime.task.event.EventCollector;
+import org.apache.iotdb.collector.runtime.task.processor.ProcessorTask;
+import org.apache.iotdb.collector.runtime.task.source.SourceTask;
+import org.apache.iotdb.commons.concurrent.IoTThreadFactory;
+import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class PullSourceTask extends SourceTask {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(PullSourceTask.class);
+
+ private static final Map<String, ExecutorService> REGISTERED_EXECUTOR_SERVICES =
+ new ConcurrentHashMap<>();
+
+ private final EventCollector processorProducer;
+ private PullSourceConsumer[] consumers;
+
+ public PullSourceTask(
+ final String taskId,
+ final Map<String, String> sourceParams,
+ final ProcessorTask processorTask) {
+ super(taskId, sourceParams, processorTask);
+ processorProducer = processorTask.makeProducer();
+ }
+
+ @Override
+ public void createInternal() throws Exception {
+ REGISTERED_EXECUTOR_SERVICES.putIfAbsent(
+ taskId,
+ new WrappedThreadPoolExecutor(
+ parallelism,
+ parallelism,
+ 0L,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(parallelism),
+ new IoTThreadFactory(taskId), // TODO: thread name
+ taskId));
+
+ final long creationTime = System.currentTimeMillis();
+ consumers = new PullSourceConsumer[parallelism];
+ for (int i = 0; i < parallelism; i++) {
+ consumers[i] =
+ new PullSourceConsumer(
+ PluginFactory.createInstance(HttpPullSource.class), processorProducer);
+ consumers[i].consumer().validate(new PipeParameterValidator(parameters));
+ consumers[i]
+ .consumer()
+ .customize(
+ parameters,
+ new CollectorSourceRuntimeConfiguration(taskId, creationTime, parallelism, i));
+ consumers[i].consumer().start();
+
+ int finalI = i;
+ REGISTERED_EXECUTOR_SERVICES
+ .get(taskId)
+ .submit(
+ () -> {
+ while (!isDropped.get()) {
+ try {
+ consumers[finalI].onScheduler();
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to pull source", e);
+ }
+
+ waitUntilRunningOrDropped();
+ }
+ });
+ }
+ }
+
+ @Override
+ public void startInternal() throws Exception {
+ // do nothing
+ }
+
+ @Override
+ public void stopInternal() throws Exception {
+ // do nothing
+ }
+
+ @Override
+ public void dropInternal() throws Exception {
+ final ExecutorService executorService = REGISTERED_EXECUTOR_SERVICES.remove(taskId);
+ if (executorService != null) {
+ executorService.shutdown();
+ }
+ }
+}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PushSourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java
similarity index 83%
rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PushSourceTask.java
rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java
index 058347e..0ea4e16 100644
--- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/source/PushSourceTask.java
+++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/source/push/PushSourceTask.java
@@ -17,11 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.collector.runtime.task.def.source;
+package org.apache.iotdb.collector.runtime.task.source.push;
import org.apache.iotdb.collector.plugin.source.HttpPushSource;
-import org.apache.iotdb.collector.runtime.task.def.processor.ProcessorTask;
-import org.apache.iotdb.collector.runtime.task.execution.EventCollector;
+import org.apache.iotdb.collector.runtime.task.event.EventCollector;
+import org.apache.iotdb.collector.runtime.task.processor.ProcessorTask;
+import org.apache.iotdb.collector.runtime.task.source.SourceTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,11 +41,7 @@ public class PushSourceTask extends SourceTask {
final ProcessorTask processorTask) {
super(taskId, sourceParams, processorTask);
- this.collector =
- new EventCollector(
- processorTask.getProcessorRingBuffer().isPresent()
- ? processorTask.getProcessorRingBuffer().get()
- : null);
+ this.collector = processorTask.makeProducer();
}
@Override