You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ka...@apache.org on 2022/06/21 03:55:03 UTC
[rocketmq-streams] 04/16: merge from snapshot-1.0.3
This is an automated email from the ASF dual-hosted git repository.
karp pushed a commit to branch snapshot-1.0.4
in repository https://gitbox.apache.org/repos/asf/rocketmq-streams.git
commit a5a6ddde987adf1143fc5a1f4ed6902dff5368aa
Merge: 93d7f775 9d3ae58b
Author: 维章 <un...@gmail.com>
AuthorDate: Mon May 23 15:32:38 2022 +0800
merge from snapshot-1.0.3
NOTICE | 2 +-
README.md | 16 +-
SUMMARY.md | 7 +
docs/README.md | 142 ------
docs/SUMMARY.md | 8 -
...225\264\344\275\223\346\236\266\346\236\204.md" | 33 --
.../2.\346\236\204\345\273\272DataStream.md" | 73 ----
.../3.\345\220\257\345\212\250DataStream.md" | 53 ---
...265\201\350\275\254\350\277\207\347\250\213.md" | 63 ---
...256\227\345\255\220\350\247\243\346\236\220.md" | 55 ---
...256\236\347\216\260\345\256\271\351\224\231.md" | 0
"docs/images/Pipeline\347\261\273\345\233\276.png" | Bin 44207 -> 0 bytes
docs/images/img.png | Bin 0 -> 38684 bytes
docs/images/img_1.png | Bin 0 -> 43711 bytes
docs/images/img_2.png | Bin 0 -> 103151 bytes
docs/images/window.png | Bin 241692 -> 0 bytes
...75\223\346\236\266\346\236\204\345\233\276.png" | Bin 60493 -> 0 bytes
...00\273\344\275\223\350\277\207\347\250\213.png" | Bin 44252 -> 0 bytes
.../\346\211\251\345\256\271\345\211\215.png" | Bin 56733 -> 0 bytes
...12\266\346\200\201\347\256\227\345\255\220.png" | Bin 35766 -> 0 bytes
"docs/images/\347\212\266\346\200\201.png" | Bin 47527 -> 0 bytes
"docs/images/\347\274\251\345\256\271.png" | Bin 51087 -> 0 bytes
docs/quick_start/README.md | 46 --
pom.xml | 118 ++---
quick_start.md | 92 ++--
.../pom.xml | 41 +-
rocketmq-streams-cep/src/test/resources/log4j.xml | 36 ++
rocketmq-streams-channel-db/pom.xml | 4 +-
.../streams/db/sink/AbstractMultiTableSink.java | 12 +-
.../streams/db/sink/DynamicMultipleDBSink.java | 11 +-
.../streams/db/sink/SelfMultiTableSink.java | 2 +-
.../streams/db/sink/SplitBySerialNumber.java | 2 +-
.../streams/db/sink/SplitByTimeMultiTableSink.java | 2 +-
rocketmq-streams-channel-es/pom.xml | 27 +-
.../{ESSinkBuilder.java => ESChannelBuilder.java} | 54 +--
.../rocketmq/streams/es/sink/ESSinkBuilder.java | 1 +
.../streams/es/sink/ESSinkOnlyChannel.java | 43 +-
.../apache/rocketmq/streams/es/sink/EsClient.java | 135 ++++++
rocketmq-streams-channel-http/pom.xml | 4 +-
rocketmq-streams-channel-kafka/pom.xml | 32 ++
.../streams/kafka/KafkaChannelBuilder.java | 52 ++-
.../apache/rocketmq/streams/kafka/KafkaSplit.java | 55 +++
.../rocketmq/streams/kafka/sink/KafkaSink.java | 200 +++++++++
.../rocketmq/streams/kafka/source/KafkaSource.java | 238 ++++++++++
.../rocketmq/streams/kafka/KafkaChannelTest.java | 104 +++++
.../src/test/resources/log4j.xml | 20 +
rocketmq-streams-channel-mqtt/pom.xml | 20 +-
.../rocketmq/streams/mqtt/source/PahoSource.java | 41 +-
rocketmq-streams-channel-rocketmq/pom.xml | 28 +-
.../apache/rocketmq/streams/debug/DebugWriter.java | 92 ++--
.../apache/rocketmq/streams/sink/RocketMQSink.java | 55 ++-
.../rocketmq/streams/RocketMQChannelTest.java | 2 +-
rocketmq-streams-channel-syslog/pom.xml | 19 +-
.../rocketmq/streams/syslog/SyslogChannel.java | 35 +-
.../streams/syslog/SyslogChannelBuilder.java | 4 +
.../streams/syslog/SyslogChannelManager.java | 6 +-
.../rocketmq/streams/syslog/SyslogServer.java | 35 +-
.../rocketmq/streams/syslog/SyslogClient.java | 22 +-
rocketmq-streams-clients/pom.xml | 14 +-
.../streams/client/source/DataStreamSource.java | 19 +
.../streams/client/transform/DataStream.java | 41 +-
.../streams/client/transform/JoinStream.java | 5 +-
.../streams/client/transform/SplitStream.java | 24 +-
.../streams/client/transform/WindowStream.java | 50 ++-
.../rocketmq/streams/client/ApplicationTest.java | 57 +++
.../rocketmq/streams/client/MqttSourceExample.java | 80 ++++
.../{sink/UserDefinedSink.java => ScriptTest.java} | 27 +-
.../apache/rocketmq/streams/client/WindowTest.java | 14 +-
.../rocketmq/streams/client/example/JoinTest.java | 7 +-
.../rocketmq/streams/client/example/SplitTest.java | 31 +-
.../streams/client/sink/UserDefinedSink.java | 2 +-
.../client/sink/UserDefinedSupportShuffleSink.java | 4 +-
rocketmq-streams-commons/pom.xml | 46 +-
.../MappedByteBufferTableWithPrimaryIndex.java | 482 +++++++++++++++++++++
.../streams/common/cache/compress/KVAddress.java | 48 +-
.../streams/common/channel/AbstractChannel.java | 14 +-
.../AbstractSupportShuffleChannelBuilder.java | 2 +-
.../common/channel/builder/IChannelBuilder.java | 25 +-
.../channel/builder/IShuffleChannelBuilder.java | 4 +-
.../common/channel/impl/CollectionSink.java | 39 +-
.../common/channel/impl/CollectionSinkBuilder.java | 34 +-
.../common/channel/impl/PrintChannelBuilder.java | 30 +-
.../channel/impl/file/FileChannelBuilder.java | 45 +-
.../streams/common/channel/impl/file/FileSink.java | 4 +-
.../common/channel/impl/file/FileSource.java | 2 +-
.../channel/impl/memory/MemoryChannelBuilder.java | 64 +++
.../common/channel/impl/memory/MemorySink.java | 4 +-
.../channel/impl/view/ViewChannelBuilder.java | 43 +-
.../streams/common/channel/impl/view/ViewSink.java | 31 +-
.../common/channel/impl/view/ViewSource.java | 66 +++
.../streams/common/channel/sink/AbstractSink.java | 17 +-
.../channel/sink/AbstractSupportShuffleSink.java | 8 +-
.../sink/AbstractSupportShuffleUDFSink.java | 6 +-
.../common/channel/sink/AbstractUDFSink.java | 23 +-
.../streams/common/channel/sink/ISink.java | 9 +-
.../impl/AbstractMultiSplitMessageCache.java | 22 +-
.../common/channel/source/AbstractSource.java | 89 +++-
.../channel/source/AbstractUnreliableSource.java | 2 +-
.../streams/common/channel/split/ISplit.java | 5 +-
.../streams/common/component/ComponentCreator.java | 4 +
.../common/configurable/AbstractConfigurable.java | 17 +-
.../common/configurable/BasedConfigurable.java | 13 +-
.../common/configurable/IConfigurableService.java | 5 +-
.../streams/common/configure/ConfigureFileKey.java | 7 +-
.../streams/common/context/AbstractContext.java | 33 +-
.../streams/common/context/MessageHeader.java | 89 ++--
.../streams/common/datatype/IntDataType.java | 3 +-
.../streams/common/datatype/ShortDataType.java | 2 +-
.../streams/common/interfaces/ISerialize.java | 9 +-
.../rocketmq/streams/common/model/NameCreator.java | 14 +-
.../NameCreatorContext.java} | 36 +-
.../common/monitor/ConsoleMonitorManager.java | 412 ++++++++++++++++++
.../streams/common/monitor/DataSyncConstants.java | 54 +++
.../streams/common/monitor/HttpClient.java | 116 +++++
.../rocketmq/streams/common/monitor/HttpUtil.java | 248 +++++++++++
.../monitor/MonitorDataSyncServiceFactory.java | 61 +++
.../common/monitor/group/MonitorCommander.java | 4 +-
.../streams/common/monitor/impl/DipperMonitor.java | 2 +-
.../streams/common/monitor/model/JobStage.java | 350 +++++++++++++++
.../streams/common/monitor/model/TraceIdsDO.java | 126 ++++++
.../common/monitor/model/TraceMonitorDO.java | 250 +++++++++++
.../service/MonitorDataSyncService.java} | 16 +-
.../service/impl/DBMonitorDataSyncImpl.java | 63 +++
.../service/impl/HttpMonitorDataSyncImpl.java | 151 +++++++
.../service/impl/RocketMQMonitorDataSyncImpl.java | 185 ++++++++
.../optimization/IHomologousOptimization.java | 2 +-
.../common/optimization/MessageGlobleTrace.java | 16 +-
.../streams/common/optimization/Re2Engine.java | 47 +-
.../common/optimization/TaskOptimization.java | 17 +-
.../optimization/fingerprint/FingerprintCache.java | 4 +-
.../optimization/fingerprint/PreFingerprint.java | 54 ++-
.../streams/common/schedule/ScheduleTask.java | 4 +-
.../common/threadpool/ThreadPoolFactory.java | 34 +-
.../AbstractMutilPipelineChainPipline.java | 81 ++--
.../streams/common/topology/ChainPipeline.java | 255 ++++++-----
.../streams/common/topology/ChainStage.java | 32 +-
.../common/topology/builder/PipelineBuilder.java | 139 ++++--
.../common/topology/metric/NotFireReason.java | 176 ++++++++
.../streams/common/topology/metric/StageGroup.java | 248 +++++++++++
.../common/topology/metric/StageMetric.java | 138 ++++++
.../common/topology/model/AbstractStage.java | 177 +++++---
.../streams/common/topology/model/IWindow.java | 3 +-
.../streams/common/topology/model/Pipeline.java | 56 ++-
.../topology/model/PipelineSourceJoiner.java | 48 --
.../topology/shuffle/IShuffleKeyGenerator.java | 11 +-
.../common/topology/shuffle/ShuffleMQCreator.java | 398 ++++++++++-------
.../topology/stages/AbstractWindowStage.java | 5 +-
.../stages/EmptyChainStage.java} | 27 +-
.../common/topology/stages/FilterChainStage.java | 157 ++-----
.../common/topology/stages/JoinChainStage.java | 3 +-
.../common/topology/stages/JoinEndChainStage.java | 11 +-
.../topology/stages/JoinStartChainStage.java | 67 +++
.../common/topology/stages/OutputChainStage.java | 81 ++--
.../topology/stages/ShuffleConsumerChainStage.java | 193 +++++++++
.../topology/stages/ShuffleProducerChainStage.java | 345 +++++++++++++++
.../topology/stages/SubPiplineChainStage.java | 138 ------
.../common/topology/stages/UnionChainStage.java | 3 +-
.../common/topology/stages/UnionEndChainStage.java | 11 +-
...onChainStage.java => UnionStartChainStage.java} | 43 +-
.../ViewChainStage.java} | 479 ++++++++++----------
.../common/topology/stages/udf/UDFChainStage.java | 27 +-
.../streams/common/topology/task/StreamsTask.java | 444 +++----------------
.../streams/common/topology/task/TaskAssigner.java | 20 +-
.../streams/common/utils/ConfigurableUtil.java | 6 +-
.../streams/common/utils/ContantsUtil.java | 33 +-
.../streams/common/utils/DataTypeUtil.java | 16 +-
.../rocketmq/streams/common/utils/FileUtil.java | 89 +++-
.../streams/common/utils/InstantiationUtil.java | 48 +-
.../streams/common/utils/JsonableUtil.java | 5 +
.../rocketmq/streams/common/utils/KryoUtil.java | 214 +++++++++
.../streams/common/utils/NameCreatorUtil.java | 60 ---
.../streams/common/utils/PipelineHTMLUtil.java | 299 +++++++++++++
.../streams/common/utils/PropertiesUtils.java | 2 +-
.../rocketmq/streams/common/utils/ReflectUtil.java | 62 ++-
.../streams/common/utils/SerializeUtil.java | 23 +-
.../streams/common/utils/ServiceLoadUtil.java | 63 +++
.../rocketmq/streams/common/utils/TraceUtil.java | 23 +-
.../rocketmq/streams/common/channel/SinkTest.java | 4 +-
rocketmq-streams-configurable/pom.xml | 9 +-
.../configurable/ConfigurableComponent.java | 17 +-
.../service/AbstractConfigurableService.java | 98 ++---
.../service/impl}/FileConfigureService.java | 2 +-
.../impl}/FileSupportParentConfigureService.java | 2 +-
.../service/impl/HttpConfigureService.java | 377 ++++++++++++++++
.../impl/HttpSupportParentConfigureService.java | 20 +-
.../service/impl}/MemoryConfigureService.java | 2 +-
.../impl}/MemorySupportParentConfigureService.java | 2 +-
rocketmq-streams-connectors/pom.xml | 9 +-
.../{IBounded.java => IScheduleCallback.java} | 9 +-
.../connectors/source/AbstractPullSource.java | 148 ++++---
.../connectors/source/MutilBatchTaskSource.java | 158 +++++++
rocketmq-streams-db-operator/pom.xml | 4 +-
.../streams/db/configuable/DBConfigureService.java | 11 +-
rocketmq-streams-dim/pom.xml | 4 +-
.../intelligence/AbstractIntelligenceCache.java | 5 +-
rocketmq-streams-examples/pom.xml | 20 +-
.../mutilconsumer/MultiStreamsExample.java | 3 +
.../streams/examples/send/ProducerFromFile.java | 8 +-
.../streams/examples/source/FileSourceExample.java | 2 +-
.../src/main/resources/joinData-2.txt | 4 -
rocketmq-streams-filter/pom.xml | 3 +-
.../streams/filter/builder/ExpressionBuilder.java | 49 +--
.../streams/filter/context/RuleContext.java | 31 +-
.../filter/engine/impl/DefaultRuleEngine.java | 29 +-
.../function/expression/CompareFunction.java | 16 +-
.../rocketmq/streams/filter/operator/Rule.java | 58 +++
.../expression/ExpressionRelationParser.java | 9 +-
.../expression/ExpressionRelationPaser.java | 107 -----
.../operator/expression/GroupExpression.java | 3 +-
.../operator/expression/RelationExpression.java | 35 +-
.../PiplineLogFingerprintAnalysis.java | 6 +-
.../dependency/BlinkRuleV2Expression.java | 5 +-
.../optimization/dependency/DependencyTree.java | 22 +-
.../dependency/SimplePipelineTree.java | 3 +-
.../dependency/StateLessDependencyTree.java | 84 ++++
.../optimization/homologous/HomologousCompute.java | 5 +-
.../homologous/HomologousOptimization.java | 13 +-
rocketmq-streams-lease/pom.xml | 4 +-
rocketmq-streams-runner/assembly/distribution.xml | 69 ---
rocketmq-streams-runner/assembly/standalone.xml | 72 ---
rocketmq-streams-runner/bin/start.sh | 58 ---
rocketmq-streams-runner/bin/stop.sh | 33 --
rocketmq-streams-runner/pom.xml | 80 ----
.../src/main/resources/log4j.xml | 51 ---
rocketmq-streams-schedule/pom.xml | 4 +-
.../schedule/job/ConfigurableExecutorJob.java | 30 +-
rocketmq-streams-script/pom.xml | 4 +-
.../function/aggregation/LastValueAccumulator.java | 67 +++
.../function/impl/distinct/DistinctFunction.java | 1 -
.../function/impl/json/JsonCreatorFunction.java | 4 +-
.../function/impl/json/UDTFFieldNameFunction.java | 50 +++
.../script/function/impl/parser/GrokFunction.java | 4 +-
.../function/impl/parser/Paser2JsonFunction.java | 17 +-
.../function/impl/parser/PaserBySplitFunction.java | 44 +-
.../function/impl/parser/RegexParserFunction.java | 24 +-
.../script/function/impl/router/RouteFunction.java | 4 +-
.../script/operator/impl/AggregationScript.java | 19 +-
.../streams/script/service/IAccumulator.java | 4 +-
.../script/service/udf/SimpleUDAFScript.java | 35 +-
.../streams/script/service/udf/UDFScript.java | 26 +-
.../streams/script/function/FunctionTest.java | 19 +-
rocketmq-streams-serviceloader/pom.xml | 4 +-
rocketmq-streams-state/pom.xml | 4 +-
.../streams/state/kv/rocksdb/RocksDBOperator.java | 37 +-
rocketmq-streams-transport-minio/pom.xml | 4 +-
rocketmq-streams-window/pom.xml | 9 +-
.../window/minibatch/MiniBatchMsgCache.java | 58 +++
.../window/minibatch/ShuffleMessageCache.java | 187 ++++++++
.../rocketmq/streams/window/model/WindowCache.java | 182 +++-----
.../streams/window/model/WindowInstance.java | 22 +-
.../window/operator/AbstractShuffleWindow.java | 18 +-
.../streams/window/operator/AbstractWindow.java | 73 +++-
.../window/operator/impl/SessionOperator.java | 2 +-
.../window/operator/impl/WindowOperator.java | 8 +-
.../streams/window/operator/join/JoinWindow.java | 3 +-
.../window/shuffle/AbstractSystemChannel.java | 94 ++--
.../streams/window/shuffle/ShuffleCache.java | 14 +-
.../streams/window/shuffle/ShuffleChannel.java | 54 +--
.../streams/window/state/impl/WindowValue.java | 54 ++-
.../streams/window/trigger/WindowTrigger.java | 1 -
.../rocketmq/streams/window/util/ShuffleUtil.java | 62 +++
.../org/apache/rocketmq/streams/RocksdbTest.java | 2 +-
docs/stream_sink/README.md => stream_sink.md | 19 +-
docs/stream_source/README.md => stream_source.md | 30 +-
.../README.md => stream_transform.md | 0
265 files changed, 9865 insertions(+), 4083 deletions(-)
diff --cc pom.xml
index 70288c26,7caa0cbc..2d139613
--- a/pom.xml
+++ b/pom.xml
@@@ -184,9 -137,9 +136,10 @@@
<exclude>**/*.xml</exclude>
<exclude>**/*.sh</exclude>
<exclude>**/*.out</exclude>
+ <exclude>**/*.sql</exclude>
<exclude>**/*.properties</exclude>
<exclude>docs/**/*</exclude>
+ <exclude>**/*.sql</exclude>
</excludes>
</configuration>
</plugin>
diff --cc rocketmq-streams-channel-es/pom.xml
index 15723b3c,fe50aeef..b4ce4d85
--- a/rocketmq-streams-channel-es/pom.xml
+++ b/rocketmq-streams-channel-es/pom.xml
@@@ -20,8 -7,8 +7,8 @@@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-streams</artifactId>
- <version>1.0.2-SNAPSHOT</version>
+ <version>1.0.2-preview-SNAPSHOT</version>
- </parent>
+ </parent>
<artifactId>rocketmq-streams-channel-es</artifactId>
<name>ROCKETMQ STREAMS :: channel-es</name>
<packaging>jar</packaging>
diff --cc rocketmq-streams-channel-kafka/pom.xml
index 00000000,89bd645c..76b73909
mode 000000,100644..100644
--- a/rocketmq-streams-channel-kafka/pom.xml
+++ b/rocketmq-streams-channel-kafka/pom.xml
@@@ -1,0 -1,32 +1,32 @@@
+ <?xml version="1.0" encoding="UTF-8"?>
+ <project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>rocketmq-streams</artifactId>
+ <groupId>org.apache.rocketmq</groupId>
- <version>1.0.2-SNAPSHOT</version>
++ <version>1.0.2-preview-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>rocketmq-streams-channel-kafka</artifactId>
+ <name>ROCKETMQ STREAMS :: channel-kafka</name>
+
+ <properties>
+ <maven.compiler.source>8</maven.compiler.source>
+ <maven.compiler.target>8</maven.compiler.target>
+ </properties>
+
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_2.12</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-streams-commons</artifactId>
+ </dependency>
+ </dependencies>
+ </project>
diff --cc rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/sink/RocketMQSink.java
index bb6e9454,d825603a..1de92e31
--- a/rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/sink/RocketMQSink.java
+++ b/rocketmq-streams-channel-rocketmq/src/main/java/org/apache/rocketmq/streams/sink/RocketMQSink.java
@@@ -17,8 -17,10 +17,11 @@@
package org.apache.rocketmq.streams.sink;
+ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collections;
+ import java.util.Comparator;
+ import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@@ -27,8 -29,10 +30,9 @@@ import java.util.Set
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
+ import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
@@@ -149,10 -154,10 +154,6 @@@ public class RocketMQSink extends Abstr
destroy();
producer = new DefaultMQProducer(groupName + "producer", true, null);
try {
-- //please not use the code，the name srv addr may be empty in jmenv
--// if (this.namesrvAddr == null || "".equals(this.namesrvAddr)) {
--// throw new RuntimeException("namesrvAddr can not be null.");
--// }
if (StringUtil.isNotEmpty(this.namesrvAddr)) {
producer.setNamesrvAddr(this.namesrvAddr);
@@@ -263,7 -275,7 +272,7 @@@
@Override
public int getSplitNum() {
- List<ISplit> splits = getSplitList();
- List<ISplit<?, ?>> splits = getSplitList();
++ List<ISplit<?,?>> splits = getSplitList();
if (splits == null || splits.size() == 0) {
return 0;
}
diff --cc rocketmq-streams-clients/pom.xml
index 8fb31672,94c7f3e7..2d15efce
--- a/rocketmq-streams-clients/pom.xml
+++ b/rocketmq-streams-clients/pom.xml
@@@ -50,6 -57,18 +57,11 @@@
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-streams-filter</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-streams-channel-syslog</artifactId>
+ </dependency>
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-streams-channel-kafka</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-streams-dbinit</artifactId>
- </dependency>
++
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-streams-window</artifactId>
diff --cc rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/source/DataStreamSource.java
index 9422519a,5004f230..1e4b15b6
--- a/rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/source/DataStreamSource.java
+++ b/rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/source/DataStreamSource.java
@@@ -178,6 -179,25 +178,25 @@@ public class DataStreamSource
return new DataStream(this.mainPipelineBuilder, this.otherPipelineBuilders, null);
}
- public DataStream fromKafka(String endpoint, String topic, String groupName) {
- return fromKafka(endpoint, topic, groupName, true);
- }
-
- public DataStream fromKafka(String endpoint, String topic, String groupName, Boolean isJson) {
- return fromKafka(endpoint, topic, groupName, isJson, 1);
- }
-
- public DataStream fromKafka(String endpoint, String topic, String groupName, Boolean isJson, int maxThread) {
- KafkaSource kafkaChannel = new KafkaSource();
- kafkaChannel.setBootstrapServers(endpoint);
- kafkaChannel.setTopic(topic);
- kafkaChannel.setGroupName(groupName);
- kafkaChannel.setJsonData(isJson);
- kafkaChannel.setMaxThread(maxThread);
- this.mainPipelineBuilder.setSource(kafkaChannel);
- return new DataStream(this.mainPipelineBuilder, null);
- }
++// public DataStream fromKafka(String endpoint, String topic, String groupName) {
++// return fromKafka(endpoint, topic, groupName, true);
++// }
++//
++// public DataStream fromKafka(String endpoint, String topic, String groupName, Boolean isJson) {
++// return fromKafka(endpoint, topic, groupName, isJson, 1);
++// }
++//
++// public DataStream fromKafka(String endpoint, String topic, String groupName, Boolean isJson, int maxThread) {
++// KafkaSource kafkaChannel = new KafkaSource();
++// kafkaChannel.setBootstrapServers(endpoint);
++// kafkaChannel.setTopic(topic);
++// kafkaChannel.setGroupName(groupName);
++// kafkaChannel.setJsonData(isJson);
++// kafkaChannel.setMaxThread(maxThread);
++// this.mainPipelineBuilder.setSource(kafkaChannel);
++// return new DataStream(this.mainPipelineBuilder, null);
++// }
+
public DataStream from(ISource<?> source) {
this.mainPipelineBuilder.setSource(source);
return new DataStream(this.mainPipelineBuilder, this.otherPipelineBuilders, null);
diff --cc rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/transform/DataStream.java
index 80ae510e,96e24a0a..563af478
--- a/rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/transform/DataStream.java
+++ b/rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/transform/DataStream.java
@@@ -51,8 -51,11 +51,10 @@@ import org.apache.rocketmq.streams.comm
import org.apache.rocketmq.streams.common.topology.ChainStage;
import org.apache.rocketmq.streams.common.topology.builder.IStageBuilder;
import org.apache.rocketmq.streams.common.topology.builder.PipelineBuilder;
-import org.apache.rocketmq.streams.common.topology.model.AbstractRule;
import org.apache.rocketmq.streams.common.topology.model.Union;
import org.apache.rocketmq.streams.common.topology.stages.FilterChainStage;
+ import org.apache.rocketmq.streams.common.topology.stages.ShuffleConsumerChainStage;
+ import org.apache.rocketmq.streams.common.topology.stages.ShuffleProducerChainStage;
import org.apache.rocketmq.streams.common.topology.stages.udf.StageBuilder;
import org.apache.rocketmq.streams.common.topology.stages.udf.UDFChainStage;
import org.apache.rocketmq.streams.common.topology.stages.udf.UDFUnionChainStage;
@@@ -646,6 -669,13 +667,13 @@@ public class DataStream implements Seri
return new DataStream(this.mainPipelineBuilder, this.otherPipelineBuilders, output);
}
- public DataStream toKafka(String bootstrapServers, String topic) {
- KafkaSink kafkaSink = new KafkaSink(bootstrapServers, topic);
- ChainStage<?> output = this.mainPipelineBuilder.createStage(kafkaSink);
- this.mainPipelineBuilder.setTopologyStages(currentChainStage, output);
- return new DataStream(this.mainPipelineBuilder, this.otherPipelineBuilders, output);
- }
++// public DataStream toKafka(String bootstrapServers, String topic) {
++// KafkaSink kafkaSink = new KafkaSink(bootstrapServers, topic);
++// ChainStage<?> output = this.mainPipelineBuilder.createStage(kafkaSink);
++// this.mainPipelineBuilder.setTopologyStages(currentChainStage, output);
++// return new DataStream(this.mainPipelineBuilder, this.otherPipelineBuilders, output);
++// }
+
public DataStream toEnhanceDBSink(String url, String userName, String password, String tableName) {
EnhanceDBSink sink = new EnhanceDBSink(url, userName, password, tableName);
ChainStage<?> output = this.mainPipelineBuilder.createStage(sink);
diff --cc rocketmq-streams-commons/pom.xml
index 5ea155d4,72af956e..2e73c6f8
--- a/rocketmq-streams-commons/pom.xml
+++ b/rocketmq-streams-commons/pom.xml
@@@ -98,11 -101,36 +101,40 @@@
<artifactId>re2j</artifactId>
<version>1.6</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-tools</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>de.ruedigermoeller</groupId>
+ <artifactId>fst</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>kryo</artifactId>
+ <version>5.3.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ <version>1.13</version>
+ </dependency>
</dependencies>
</project>
diff --cc rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java
index 939b1d54,320cbd12..2bc7e55f
--- a/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java
+++ b/rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/source/AbstractSource.java
@@@ -18,9 -18,10 +18,11 @@@ package org.apache.rocketmq.streams.com
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
+
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
+ import java.util.Arrays;
+ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@@ -58,15 -58,22 +60,23 @@@ import org.apache.rocketmq.streams.comm
public abstract class AbstractSource extends BasedConfigurable implements ISource<AbstractSource>, ILifeCycle {
public static String CHARSET = "UTF-8";
+ /**
+ * 输入的消息是否为json
+ */
+ protected Boolean isJsonData = true;
+ /**
+ * 输入的消息是否为json array
+ */
+ protected Boolean msgIsJsonArray = false;
- protected Boolean isJsonData = true;//输入的消息是否为json
- protected Boolean msgIsJsonArray = false;//输入的消息是否为json array
@ENVDependence
- protected String groupName;//group name
+ protected String groupName;
+
protected int maxThread = Runtime.getRuntime().availableProcessors();
+
@ENVDependence
protected String topic = "";
+ protected String namesrvAddr;
/**
* 多长时间做一次checkpoint
*/
@@@ -136,11 -167,9 +170,10 @@@
* @param message
* @return
*/
- public AbstractContext doReceiveMessage(JSONObject message, boolean needSetCheckPoint, String queueId, String offset) {
+ public AbstractContext doReceiveMessage(JSONObject message, boolean needSetCheckPoint, String queueId,
+ String offset) {
Message msg = createMessage(message, queueId, offset, needSetCheckPoint);
- AbstractContext context = executeMessage(msg);
- return context;
+ return executeMessage(msg);
}
/**
diff --cc rocketmq-streams-connectors/pom.xml
index 8c8277e3,d544e3d5..e557a19a
mode 100755,100644..100644
--- a/rocketmq-streams-connectors/pom.xml
+++ b/rocketmq-streams-connectors/pom.xml
diff --cc rocketmq-streams-examples/pom.xml
index 62a4038c,da6a130b..7ced0cb0
--- a/rocketmq-streams-examples/pom.xml
+++ b/rocketmq-streams-examples/pom.xml
@@@ -39,6 -41,6 +41,22 @@@
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-streams-clients</artifactId>
</dependency>
++ <dependency>
++ <groupId>com.alibaba</groupId>
++ <artifactId>fastjson</artifactId>
++ </dependency>
++ <dependency>
++ <groupId>junit</groupId>
++ <artifactId>junit</artifactId>
++ </dependency>
++ <dependency>
++ <groupId>org.apache.rocketmq</groupId>
++ <artifactId>rocketmq-client</artifactId>
++ </dependency>
++ <dependency>
++ <groupId>org.apache.rocketmq</groupId>
++ <artifactId>rocketmq-streams-commons</artifactId>
++ </dependency>
</dependencies>
<packaging>jar</packaging>
diff --cc rocketmq-streams-examples/src/main/java/org/apache/rocketmq/streams/examples/mutilconsumer/MultiStreamsExample.java
index f5f3d46c,2b823e44..8835e943
--- a/rocketmq-streams-examples/src/main/java/org/apache/rocketmq/streams/examples/mutilconsumer/MultiStreamsExample.java
+++ b/rocketmq-streams-examples/src/main/java/org/apache/rocketmq/streams/examples/mutilconsumer/MultiStreamsExample.java
@@@ -24,13 -23,11 +24,15 @@@ import com.alibaba.fastjson.JSONObject
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+
++import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.streams.client.StreamBuilder;
import org.apache.rocketmq.streams.client.source.DataStreamSource;
import org.apache.rocketmq.streams.client.strategy.WindowStrategy;
++import org.apache.rocketmq.streams.client.transform.DataStream;
import org.apache.rocketmq.streams.client.transform.window.Time;
import org.apache.rocketmq.streams.client.transform.window.TumblingWindow;
+import org.apache.rocketmq.streams.examples.send.ProducerFromFile;
import static org.apache.rocketmq.streams.examples.aggregate.Constant.NAMESRV_ADDRESS;
import static org.apache.rocketmq.streams.examples.aggregate.Constant.RMQ_CONSUMER_GROUP_NAME;
@@@ -64,6 -61,6 +66,7 @@@ public class MultiStreamsExample
private static void runOneStreamsClient(int index) {
DataStreamSource source = StreamBuilder.dataStream("namespace" + index, "pipeline" + index);
++
source.fromRocketmq(
RMQ_TOPIC,
RMQ_CONSUMER_GROUP_NAME,
diff --cc rocketmq-streams-examples/src/main/java/org/apache/rocketmq/streams/examples/send/ProducerFromFile.java
index 163d8116,58d3710c..0508c303
--- a/rocketmq-streams-examples/src/main/java/org/apache/rocketmq/streams/examples/send/ProducerFromFile.java
+++ b/rocketmq-streams-examples/src/main/java/org/apache/rocketmq/streams/examples/send/ProducerFromFile.java
@@@ -17,8 -17,8 +17,12 @@@
*
*/
-package org.apache.rocketmq.streams.examples.aggregate;
+package org.apache.rocketmq.streams.examples.send;
+
++import org.apache.rocketmq.client.producer.DefaultMQProducer;
++import org.apache.rocketmq.common.message.Message;
++import org.apache.rocketmq.remoting.common.RemotingHelper;
+
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
@@@ -26,44 -26,18 +30,40 @@@ import java.io.IOException
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.client.producer.SendResult;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.remoting.common.RemotingHelper;
+import java.util.concurrent.atomic.AtomicLong;
- import org.apache.rocketmq.client.producer.DefaultMQProducer;
- import org.apache.rocketmq.common.message.Message;
- import org.apache.rocketmq.remoting.common.RemotingHelper;
-
public class ProducerFromFile {
+ private static final DefaultMQProducer producer = new DefaultMQProducer("test-group");
+ private static final AtomicLong count = new AtomicLong(0);
+ private static boolean init = false;
- public static void produce(String filePath, String nameServ, String topic) {
- try {
- DefaultMQProducer producer = new DefaultMQProducer("test-group");
+ private static synchronized void initProducer(String nameServ) throws Throwable {
+ if (!init) {
producer.setNamesrvAddr(nameServ);
producer.start();
+ init = true;
+ }
+ }
+
+ public static void produceInLoop(String filePath, String nameServ, String topic, long interval) {
+ while (true) {
+ try {
+ produce(filePath, nameServ, topic, false);
+
+ Thread.sleep(interval);
+
+ if (count.get() % 500 == 0) {
+ System.out.println("send message num: " + count.get());
+ }
+ } catch (Throwable t) {
+ t.printStackTrace();
+ }
+ }
+ }
+
+ public static void produce(String filePath, String nameServ, String topic, boolean shutdown) {
+ try {
+ initProducer(nameServ);
List<String> result = ProducerFromFile.read(filePath);
diff --cc rocketmq-streams-window/pom.xml
index 40fcbf94,7a2be2dd..3ce4dac8
--- a/rocketmq-streams-window/pom.xml
+++ b/rocketmq-streams-window/pom.xml
@@@ -48,10 -50,6 +50,13 @@@
<groupId>org.rocksdb</groupId>
<artifactId>rocksdbjni</artifactId>
</dependency>
-
-
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-tools</artifactId>
+ </dependency>
++ <dependency>
++ <groupId>org.apache.rocketmq</groupId>
++ <artifactId>rocketmq-streams-commons</artifactId>
++ </dependency>
</dependencies>
</project>
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/minibatch/MiniBatchMsgCache.java
index 00000000,39629cc1..a36309e2
mode 000000,100644..100644
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/minibatch/MiniBatchMsgCache.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/minibatch/MiniBatchMsgCache.java
@@@ -1,0 -1,76 +1,58 @@@
+ /*
+ * 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.rocketmq.streams.window.minibatch;
+
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+ import org.apache.commons.lang3.tuple.Pair;
+ import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageFlushCallBack;
+ import org.apache.rocketmq.streams.common.channel.sinkcache.impl.AbstractMultiSplitMessageCache;
+ import org.apache.rocketmq.streams.common.channel.sinkcache.impl.MessageCache;
+ import org.apache.rocketmq.streams.common.channel.split.ISplit;
+ import org.apache.rocketmq.streams.common.context.IMessage;
-import org.apache.rocketmq.streams.common.context.Message;
-import org.apache.rocketmq.streams.common.context.MessageHeader;
+ import org.apache.rocketmq.streams.common.topology.shuffle.IShuffleKeyGenerator;
-import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
-import org.apache.rocketmq.streams.common.utils.ReflectUtil;
-import org.apache.rocketmq.streams.common.utils.StringUtil;
-import org.apache.rocketmq.streams.script.operator.impl.AggregationScript;
-import org.apache.rocketmq.streams.window.model.WindowInstance;
+ import org.apache.rocketmq.streams.window.operator.AbstractShuffleWindow;
-import org.apache.rocketmq.streams.window.operator.AbstractWindow;
-import org.apache.rocketmq.streams.window.state.impl.WindowValue;
-import org.apache.rocketmq.streams.window.util.ShuffleUtil;
+
+ public class MiniBatchMsgCache extends AbstractMultiSplitMessageCache<Pair<ISplit,IMessage>> {
+ public static String SHUFFLE_KEY="shuffle_key";
+
+
+
+ protected transient IShuffleKeyGenerator shuffleKeyGenerator;
+ protected transient AbstractShuffleWindow window;
+
+
+
+
+ public MiniBatchMsgCache(
+ IMessageFlushCallBack<Pair<ISplit,IMessage>> flushCallBack, IShuffleKeyGenerator shuffleKeyGenerator,
+ AbstractShuffleWindow window) {
+ super(flushCallBack);
+ this.shuffleKeyGenerator=shuffleKeyGenerator;
+ this.window=window;
+ }
+
+
+ @Override protected String createSplitId(Pair<ISplit, IMessage> msg) {
+ return msg.getLeft().getQueueId();
+ }
+
+ @Override protected MessageCache createMessageCache() {
+ ShuffleMessageCache messageCache=new ShuffleMessageCache(this.flushCallBack);
+ messageCache.setWindow(window);
+ messageCache.setShuffleKeyGenerator(shuffleKeyGenerator);
+ return messageCache;
+ }
+ }
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/model/WindowInstance.java
index c749bd84,fed18863..4062b1f5
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/model/WindowInstance.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/model/WindowInstance.java
@@@ -116,6 -122,19 +116,10 @@@ public class WindowInstance extends Ent
return windowInstance;
}
- /**
- * 创建window instance的唯一ID
- *
- * @return
- */
+ public String createWindowInstanceId() {
+ return MapKeyUtil.createKey(splitId, windowNameSpace, windowName, windowInstanceName, startTime, endTime);
+ }
+
- public String createWindowInstanceIdWithoutSplitid() {
- return MapKeyUtil.createKey(windowNameSpace, windowName, windowInstanceName, startTime, endTime);
- }
-
public String createWindowInstanceTriggerId() {
return MapKeyUtil.createKey(splitId, windowNameSpace, windowName, windowInstanceName, startTime, endTime, fireTime);
}
@@@ -162,7 -254,20 +165,18 @@@
* @return
* @Param isWindowInstance2DB 如果是秒级窗口，可能windowinstacne不必存表，只在内存保存，可以通过这个标志设置
*/
- public static List<WindowInstance> getOrCreateWindowInstance(AbstractWindow window, Long occurTime,
- int timeUnitAdjust, String queueId) {
+ public static List<WindowInstance> getOrCreateWindowInstance(AbstractWindow window, Long occurTime, int timeUnitAdjust, String queueId) {
+ return getOrCreateWindowInstance(window,occurTime,timeUnitAdjust,queueId,false);
+ }
+ /**
+ * 查询或者创建Window的实例，滑动窗口有可能返回多个，滚动窗口返回一个
+ *
+ * @param window
+ * @param occurTime
+ * @return
+ * @Param isWindowInstance2DB 如果是秒级窗口，可能windowinstacne不必存表，只在内存保存，可以通过这个标志设置
+ */
- public static List<WindowInstance> getOrCreateWindowInstance(AbstractWindow window, Long occurTime,
- int timeUnitAdjust, String queueId, boolean isCreateOnly) {
++ public static List<WindowInstance> getOrCreateWindowInstance(AbstractWindow window, Long occurTime, int timeUnitAdjust, String queueId, boolean isCreateOnly) {
int windowSlideInterval = window.getSlideInterval();
int windowSizeInterval = window.getSizeInterval();
if (windowSlideInterval == 0) {
@@@ -246,10 -345,9 +264,10 @@@
}
}
List<WindowInstance> lostInstanceList = null;
+ //todo 这里针对lost的都创建一次
lostInstanceList = WindowInstance.createWindowInstances(window, lostWindowTimeList, lostFireList, queueId);
instanceList.addAll(lostInstanceList);
- if (CollectionUtil.isNotEmpty(lostInstanceList)) {
+ if (CollectionUtil.isNotEmpty(lostInstanceList)&&!isCreateOnly) {
for (WindowInstance windowInstance : instanceList) {
List<WindowInstance> emitInstances = createEmitWindowInstance(window, windowInstance);
if (emitInstances != null && emitInstances.size() > 0) {
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/AbstractShuffleWindow.java
index 070a40f0,5338734f..7c78f478
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/AbstractShuffleWindow.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/AbstractShuffleWindow.java
@@@ -85,8 -74,7 +100,7 @@@ public abstract class AbstractShuffleWi
Set<String> splitIds = new HashSet<>();
splitIds.add(windowInstance.getSplitId());
shuffleChannel.flush(splitIds);
-
- return fireWindowInstance(windowInstance, windowInstance.getSplitId(), queueId2Offset);
+ return doFireWindowInstance(windowInstance);
}
/**
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/AbstractWindow.java
index 49d79c5b,8fd07737..eb548aa6
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/AbstractWindow.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/AbstractWindow.java
@@@ -17,6 -17,17 +17,8 @@@
package org.apache.rocketmq.streams.window.operator;
import com.alibaba.fastjson.JSONObject;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
++
+ import javafx.util.Pair;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@@ -137,8 -164,8 +145,9 @@@ public abstract class AbstractWindow ex
protected boolean isLocalStorageOnly = true;//是否只用本地存储，可以提高性能，但不保证可靠性
protected String reduceSerializeValue;//用户自定义的operator的序列化字节数组，做了base64解码
protected transient IReducer reducer;
-
+ protected transient Long maxPartitionNum = 100000000L;
+ protected String mapFunctionSerializeValue;//用户自定义的operator的序列化字节数组，做了base64解码
+ protected transient MapFunction<JSONObject, Pair<WindowInstance, JSONObject>> mapFunction;
/**
* the computed column and it's process of computing
*/
@@@ -164,12 -191,23 +173,13 @@@
*/
protected transient String WINDOW_NAME;
- /**
- * 内部使用,定期检查窗口有没有触发
- */
- //protected transient ScheduledExecutorService fireWindowInstanceChecker =new ScheduledThreadPoolExecutor(3);
-
- // protected transient ExecutorService deleteService = Executors.newSingleThreadExecutor();
protected volatile transient WindowCache windowCache;
- protected transient WindowStorage storage;
+ protected transient IStorage storage;
protected transient WindowTrigger windowFireSource;
- protected transient SQLCache sqlCache;
protected transient EventTimeManager eventTimeManager;
+ protected transient ISink contextMsgSink;
- //create and save window instacne max partitionNum and window max eventTime
- protected transient IWindowMaxValueManager windowMaxValueManager;
-
public AbstractWindow() {
setType(IWindow.TYPE);
}
@@@ -199,8 -242,12 +209,12 @@@
byte[] bytes = Base64Utils.decode(this.reduceSerializeValue);
reducer = InstantiationUtil.deserializeObject(bytes);
}
+ if (StringUtil.isNotEmpty(this.mapFunctionSerializeValue)) {
+ byte[] bytes = Base64Utils.decode(this.mapFunctionSerializeValue);
+ this.mapFunction = InstantiationUtil.deserializeObject(bytes);
+ }
eventTimeManager = new EventTimeManager();
- windowMaxValueManager = new WindowMaxValueManager(this, sqlCache);
+
return success;
}
@@@ -306,11 -347,11 +320,12 @@@
* @param message
* @return
*/
- protected String generateShuffleKey(IMessage message) {
+ @Override
+ public String generateShuffleKey(IMessage message) {
if (StringUtil.isEmpty(groupByFieldName)) {
- return null;
+ return "globle_window";
}
+
JSONObject msg = message.getMessageBody();
String[] fieldNames = groupByFieldName.split(";");
String[] values = new String[fieldNames.length];
@@@ -773,13 -845,45 +796,53 @@@
this.maxDelay = maxDelay;
}
+ public Long getMaxPartitionNum() {
+ return maxPartitionNum;
+ }
+
+ public void setMaxPartitionNum(Long maxPartitionNum) {
+ this.maxPartitionNum = maxPartitionNum;
+ }
+
public abstract boolean supportBatchMsgFinish();
+
+ public ISink getContextMsgSink() {
+ return contextMsgSink;
+ }
+
+ public String getContextMsgSinkName() {
+ return contextMsgSinkName;
+ }
+
+ public void setContextMsgSinkName(String contextMsgSinkName) {
+ this.contextMsgSinkName = contextMsgSinkName;
+ }
+
+ public String getMapFunctionSerializeValue() {
+ return mapFunctionSerializeValue;
+ }
+
+ public void setMapFunctionSerializeValue(String mapFunctionSerializeValue) {
+ this.mapFunctionSerializeValue = mapFunctionSerializeValue;
+ }
+
+ public void saveMsgContext(String queueId,WindowInstance windowInstance, List<IMessage> messages) {
+ if(this.mapFunction!=null&&this.contextMsgSink!=null){
+ if(messages!=null){
+ for(IMessage message:messages){
+ JSONObject msg=message.getMessageBody();
+ try {
+ msg=this.mapFunction.map(new Pair(windowInstance,msg));
+ Message copyMsg=new Message(msg);
+ copyMsg.getHeader().setQueueId(queueId);
+ copyMsg.getHeader().setOffset(message.getHeader().getOffset());
+ this.contextMsgSink.batchAdd(copyMsg);
+ } catch (Exception e) {
+ throw new RuntimeException("save window context msg error ",e);
+ }
+ }
+ this.contextMsgSink.flush();
+ }
+ }
+ }
}
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/SessionOperator.java
index 65ac261b,8da03987..0c10a9bf
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/SessionOperator.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/SessionOperator.java
@@@ -515,13 -481,12 +515,13 @@@ public class SessionOperator extends Wi
store(lastValueMap, instance, queueId);
}
+
@Override
public long incrementAndGetSplitNumber(WindowInstance instance, String shuffleId) {
- long number = super.incrementAndGetSplitNumber(instance, shuffleId);
- if (number > 900000000) {
- this.getWindowMaxValueManager().resetSplitNum(instance, shuffleId);
+ long numer = super.incrementAndGetSplitNumber(instance, shuffleId);
+ if (numer > 900000000) {
+ this.storage.putMaxPartitionNum(shuffleId, instance.getWindowInstanceId(), numer);
}
- return number;
+ return numer;
}
- }
+ }
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/WindowOperator.java
index 65f326f6,d2806081..de6dce22
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/WindowOperator.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/impl/WindowOperator.java
@@@ -16,113 -16,155 +16,119 @@@
*/
package org.apache.rocketmq.streams.window.operator.impl;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import org.apache.rocketmq.streams.common.channel.split.ISplit;
++
import org.apache.rocketmq.streams.common.context.IMessage;
import org.apache.rocketmq.streams.common.context.MessageOffset;
-import org.apache.rocketmq.streams.common.utils.CollectionUtil;
-import org.apache.rocketmq.streams.common.utils.DateUtil;
import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
import org.apache.rocketmq.streams.common.utils.StringUtil;
-import org.apache.rocketmq.streams.db.driver.batchloader.IRowOperator;
-import org.apache.rocketmq.streams.db.driver.orm.ORMUtil;
-import org.apache.rocketmq.streams.script.operator.impl.AggregationScript;
import org.apache.rocketmq.streams.window.debug.DebugWriter;
-import org.apache.rocketmq.streams.window.model.FunctionExecutor;
import org.apache.rocketmq.streams.window.model.WindowInstance;
import org.apache.rocketmq.streams.window.operator.AbstractShuffleWindow;
-import org.apache.rocketmq.streams.window.operator.AbstractWindow;
-import org.apache.rocketmq.streams.window.sqlcache.impl.FiredNotifySQLElement;
import org.apache.rocketmq.streams.window.state.WindowBaseValue;
import org.apache.rocketmq.streams.window.state.impl.WindowValue;
-import org.apache.rocketmq.streams.window.storage.IWindowStorage;
-import org.apache.rocketmq.streams.window.storage.ShufflePartitionManager;
-import org.apache.rocketmq.streams.window.storage.WindowStorage.WindowBaseValueIterator;
-
-public class WindowOperator extends AbstractShuffleWindow {
+import org.apache.rocketmq.streams.window.storage.IteratorWrap;
+import org.apache.rocketmq.streams.window.storage.RocksdbIterator;
+import org.apache.rocketmq.streams.window.storage.WindowType;
- import java.util.*;
- private static final String ORDER_BY_SPLIT_NUM = "_order_by_split_num_";//key=_order;queueid,windowinstanceid,partitionNum
++import java.util.ArrayList;
++import java.util.Comparator;
++import java.util.HashMap;
++import java.util.List;
++import java.util.Map;
+import java.util.Map.Entry;
+import java.util.stream.Collectors;
+public class WindowOperator extends AbstractShuffleWindow {
public WindowOperator() {
super();
}
- protected transient boolean supportQuickStoreModel=false;
- protected transient List<String> schema=new ArrayList<>();
+ @Override
+ public int doFireWindowInstance(WindowInstance instance) {
+ String windowInstanceId = instance.getWindowInstanceId();
+ String queueId = instance.getSplitId();
- @Deprecated
- public WindowOperator(String timeFieldName, int windowPeriodMinute) {
- super();
- super.timeFieldName = timeFieldName;
- super.sizeInterval = windowPeriodMinute;
- }
+ RocksdbIterator<WindowBaseValue> rocksdbIterator = storage.getWindowBaseValue(queueId, windowInstanceId, WindowType.NORMAL_WINDOW, null);
- @Deprecated
- public WindowOperator(String timeFieldName, int windowPeriodMinute, String calFieldName) {
- super();
- super.timeFieldName = timeFieldName;
- super.sizeInterval = windowPeriodMinute;
- }
+ ArrayList<WindowValue> windowValues = new ArrayList<>();
+ while (rocksdbIterator.hasNext()) {
+ IteratorWrap<WindowBaseValue> next = rocksdbIterator.next();
+ WindowValue data = (WindowValue)next.getData();
+ windowValues.add(data);
+ }
+
+ windowValues.sort(Comparator.comparingLong(WindowBaseValue::getPartitionNum));
- public WindowOperator(int sizeInterval, String groupByFieldName, Map<String, String> select) {
- this.sizeInterval = sizeInterval;
- this.slideInterval = sizeInterval;
- this.groupByFieldName = groupByFieldName;
- this.setSelectMap(select);
+ int fireCount = sendBatch(windowValues, queueId, 0);
+
+ clearFire(instance);
+
+ return fireCount;
}
- protected transient AtomicInteger shuffleCount = new AtomicInteger(0);
- protected transient AtomicInteger fireCountAccumulator = new AtomicInteger(0);
- protected transient AtomicLong fireCost=new AtomicLong(0);
- @Override
- public int fireWindowInstance(WindowInstance instance, String queueId, Map<String, String> queueId2Offset) {
- List<WindowValue> windowValues = new ArrayList<>();
- int fireCount = 0;
- //long startTime = System.currentTimeMillis();
- //int sendCost = 0;
- // int currentCount = 0;
- //for(String queueId:currentQueueIds){
- WindowBaseValueIterator<WindowBaseValue> it = storage.loadWindowInstanceSplitData(getOrderBypPrefix(), queueId, instance.createWindowInstanceId(), null, getWindowBaseValueClass());
- if (queueId2Offset != null) {
- String offset = queueId2Offset.get(queueId);
- if (StringUtil.isNotEmpty(offset)) {
- it.setPartitionNum(Long.valueOf(offset));
- }
+ private int sendBatch(List<WindowValue> windowValues, String queueId, int fireCount) {
+ if (windowValues == null || windowValues.size() == 0) {
+ return fireCount;
}
- while (it.hasNext()) {
- WindowBaseValue windowBaseValue = it.next();
- if (windowBaseValue == null) {
- continue;
- }
- WindowValue windowValue = (WindowValue) windowBaseValue;
-
- Integer currentValue = getValue(windowValue, "total");
-
- fireCountAccumulator.addAndGet(currentValue);
- windowValues.add((WindowValue) windowBaseValue);
- if (windowValues.size() >= windowCache.getBatchSize()) {
- long sendFireCost = System.currentTimeMillis();
- sendFireMessage(windowValues, queueId);
- //sendCost += (System.currentTimeMillis() - sendFireCost);
- fireCount += windowValues.size();
- windowValues = new ArrayList<>();
- }
- }
- if (windowValues.size() > 0) {
- long sendFireCost = System.currentTimeMillis();
+ if (windowValues.size() <= windowCache.getBatchSize()) {
sendFireMessage(windowValues, queueId);
- // sendCost += (System.currentTimeMillis() - sendFireCost);
+
fireCount += windowValues.size();
+
+ return fireCount;
+ } else {
+ ArrayList<WindowValue> temp = new ArrayList<>();
+ for (int i = 0; i < windowCache.getBatchSize(); i++) {
+ temp.add(windowValues.remove(i));
+ }
+
+ sendFireMessage(temp, queueId);
+
+ return sendBatch(windowValues, queueId, fireCount + windowCache.getBatchSize());
}
- clearFire(instance);
- this.sqlCache.addCache(new FiredNotifySQLElement(queueId, instance.createWindowInstanceId()));
- //long cost= this.fireCost.addAndGet(System.currentTimeMillis()-startTime);
- // System.out.println("fire cost is "+cost+" "+ DateUtil.getCurrentTimeString());
- return fireCount;
}
- protected transient Map<String, Integer> shuffleWindowInstanceId2MsgCount = new HashMap<>();
- protected transient int windowvaluecount = 0;
- protected transient AtomicLong shuffleCost=new AtomicLong(0);
+
@Override
public void shuffleCalculate(List<IMessage> messages, WindowInstance instance, String queueId) {
+ Long startTime=System.currentTimeMillis();
DebugWriter.getDebugWriter(getConfigureName()).writeShuffleCalcultateReceveMessage(instance, messages, queueId);
+
List<String> sortKeys = new ArrayList<>();
Map<String, List<IMessage>> groupBy = groupByGroupName(messages, sortKeys);
- Set<String> groupByKeys = groupBy.keySet();
- List<String> storeKeys = new ArrayList<>();
- for (String groupByKey : groupByKeys) {
- String storeKey = createStoreKey(queueId, groupByKey, instance);
- storeKeys.add(storeKey);
+
+ RocksdbIterator<WindowBaseValue> windowBaseValue = storage.getWindowBaseValue(queueId, instance.getWindowInstanceId(), WindowType.NORMAL_WINDOW, null);
+
+ ArrayList<WindowBaseValue> windowValues = new ArrayList<>();
+ while (windowBaseValue.hasNext()) {
+ IteratorWrap<WindowBaseValue> next = windowBaseValue.next();
+ windowValues.add(next.getData());
}
- Map<String, WindowBaseValue> allWindowValues = new HashMap<>();
- //从存储中，查找window value对象，value是对象的json格式
- Map<String, WindowBaseValue> existWindowValues = storage.multiGet(getWindowBaseValueClass(), storeKeys, instance.createWindowInstanceId(), queueId);
- // Iterator<Entry<String, List<IMessage>>> it = groupBy.entrySet().iterator();
- for (String groupByKey : sortKeys) {
+ Map<String, List<WindowValue>> temp = windowValues.stream().map((value) -> (WindowValue) value).collect(Collectors.groupingBy(WindowValue::getMsgKey));
+
+ Map<String, List<WindowValue>> groupByMsgKey = new HashMap<>(temp);
+
+ List<WindowValue> allWindowValues = new ArrayList<>();
+
+ //处理不同groupBy的message
+ for (String groupByKey : sortKeys) {
List<IMessage> msgs = groupBy.get(groupByKey);
String storeKey = createStoreKey(queueId, groupByKey, instance);
- WindowValue windowValue = (WindowValue) existWindowValues.get(storeKey);
- ;
- if (windowValue == null) {
- windowvaluecount++;
+
+ //msgKey 为唯一键
+ List<WindowValue> windowValueList = groupByMsgKey.get(storeKey);
+ WindowValue windowValue;
+ if (windowValueList == null || windowValueList.size() == 0) {
windowValue = createWindowValue(queueId, groupByKey, instance);
- // windowValue.setOrigOffset(msgs.get(0).getHeader().getOffset());
+ } else {
+ windowValue = windowValueList.get(0);
}
- allWindowValues.put(storeKey, windowValue);
- windowValue.incrementUpdateVersion();
- Integer origValue = getValue(windowValue, "total");
+ allWindowValues.add(windowValue);
+ windowValue.incrementUpdateVersion();
if (msgs != null) {
for (IMessage message : msgs) {
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java
index 9c1c0af6,5eb3b784..f356a286
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/operator/join/JoinWindow.java
@@@ -60,11 -62,16 +60,10 @@@ public class JoinWindow extends Abstrac
protected String joinType;//join类型，值为INNER,LEFT
protected String expression;//条件表达式。在存在非等值比较时使用
- protected transient DBOperator joinOperator = new DBOperator();
- protected String rightDependentTableName;
- // @Override
- // protected void addPropertyToMessage(IMessage oriMessage, JSONObject oriJson){
- // oriJson.put("AbstractWindow", this);
- //
- // }
-
@Override
- protected int fireWindowInstance(WindowInstance instance, String shuffleId, Map<String, String> queueId2Offsets) {
+ protected int doFireWindowInstance(WindowInstance instance) {
+ //todo 只是清理吗？
clearFire(instance);
return 0;
}
@@@ -359,8 -422,18 +358,8 @@@
}
@Override
- protected String generateShuffleKey(IMessage message) {
+ public String generateShuffleKey(IMessage message) {
- String routeLabel =null;
- String lable = message.getHeader().getMsgRouteFromLable();
- if (lable != null) {
- if (lable.equals(rightDependentTableName)) {
- routeLabel = MessageHeader.JOIN_RIGHT;
- } else {
- routeLabel = MessageHeader.JOIN_LEFT;
- }
- } else {
- throw new RuntimeException("can not dipatch message, need route label " + toJson());
- }
+ String routeLabel = message.getHeader().getMsgRouteFromLable();
String messageKey = generateKey(message.getMessageBody(), routeLabel, leftJoinFieldNames, rightJoinFieldNames);
return messageKey;
}
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/AbstractSystemChannel.java
index 6c4659b3,d0e499f2..0a3845fd
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/AbstractSystemChannel.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/AbstractSystemChannel.java
@@@ -54,11 -55,12 +55,12 @@@ public abstract class AbstractSystemCha
protected static final String CHANNEL_PROPERTY_KEY_PREFIX = "CHANNEL_PROPERTY_KEY_PREFIX";
protected static final String CHANNEL_TYPE = "CHANNEL_TYPE";
- protected ISource consumer;
+ protected ISource<?> consumer;
protected AbstractSupportShuffleSink producer;
protected Map<String, String> channelConfig = new HashMap<>();
- protected volatile boolean hasCreateShuffleChannel = false;
+ protected boolean hasCreateShuffleChannel = false;
+ protected transient AtomicBoolean hasStart = new AtomicBoolean(false);
public void startChannel() {
if (consumer == null) {
return;
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/ShuffleCache.java
index 5139f707,054a418d..7b4d8b4f
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/ShuffleCache.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/ShuffleCache.java
@@@ -22,11 -22,11 +22,11 @@@ import java.util.Collections
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.Future;
-
import org.apache.commons.lang3.tuple.Pair;
+ import org.apache.rocketmq.streams.common.channel.sink.AbstractSink;
import org.apache.rocketmq.streams.common.context.IMessage;
-import org.apache.rocketmq.streams.db.driver.orm.ORMUtil;
+import org.apache.rocketmq.streams.common.context.MessageOffset;
import org.apache.rocketmq.streams.window.debug.DebugWriter;
import org.apache.rocketmq.streams.window.model.WindowCache;
import org.apache.rocketmq.streams.window.model.WindowInstance;
@@@ -36,9 -37,8 +36,9 @@@ import org.apache.rocketmq.streams.wind
/**
* save receiver messages into cachefilter when checkpoint/autoflush/flush， process cachefilter message
*/
- public class ShuffleCache extends WindowCache {
+ public class ShuffleCache extends AbstractSink {
protected AbstractShuffleWindow window;
+ private HashMap<String, Boolean> hasLoad = new HashMap<>();
public ShuffleCache(AbstractShuffleWindow window) {
this.window = window;
@@@ -64,64 -54,23 +64,65 @@@
for (Pair<String, String> queueIdAndInstanceKey : keys) {
String queueId = queueIdAndInstanceKey.getLeft();
String windowInstanceId = queueIdAndInstanceKey.getRight();
+
List<IMessage> messages = instance2Messages.get(queueIdAndInstanceKey);
+
WindowInstance windowInstance = windowInstanceMap.get(windowInstanceId);
+
DebugWriter.getDebugWriter(window.getConfigureName()).writeShuffleReceive(window, messages, windowInstance);
+
+ stateMustLoad(queueId);
+
window.shuffleCalculate(messages, windowInstance, queueId);
+
+ //保存处理进度
saveSplitProgress(queueId, messages);
+ window.saveMsgContext(queueId,windowInstance,messages);
}
- return true;
+ return true;
}
+ private void stateMustLoad(String queueId) {
+ Boolean load = this.hasLoad.get(queueId);
+ if (load != null && load) {
+ return;
+ }
+
+ //在计算之前需要异步加载状态完成
+ HashMap<String, Future<?>> loadResult = this.window.getShuffleChannel().getLoadResult();
+ Future<?> future = loadResult.get(queueId);
+
+ if (future == null) {
+ return;
+ }
+
+ try {
+ long before = System.currentTimeMillis();
+ future.get();
+ long after = System.currentTimeMillis();
+
+ System.out.println("message wait before state recover:[" + (after - before) + "] ms, queueId=" + queueId);
+
+ for (String loadQueueId : loadResult.keySet()) {
+ hasLoad.put(loadQueueId, true);
+ }
+ } catch (Throwable t) {
+ throw new RuntimeException("check remote with queueId:" + queueId + ",error", t);
+ }
+ }
+
/**
- * save consumer progress（offset）for groupby source queueId
+ * save consumer progress（offset）for groupby source shuffleId
+ * window configName: name_window_10001
+ * shuffleId: shuffle_NormalTestTopic_namespace_name_broker-a_001
+ * oriQueueId: NormalTestTopic2_broker-a_000
*
- * @param queueId
+ * @param shuffleId
* @param messages
*/
- protected void saveSplitProgress(String queueId, List<IMessage> messages) {
+ protected void saveSplitProgress(String shuffleId, List<IMessage> messages) {
+ IStorage delegator = this.window.getStorage();
+
Map<String, String> queueId2OrigOffset = new HashMap<>();
Boolean isLong = false;
for (IMessage message : messages) {
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/ShuffleChannel.java
index f9bd3413,9280a2e4..d834af41
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/ShuffleChannel.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/shuffle/ShuffleChannel.java
@@@ -18,7 -18,18 +18,8 @@@ package org.apache.rocketmq.streams.win
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang3.StringUtils;
+ import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.rocketmq.streams.common.batchsystem.BatchFinishMessage;
@@@ -42,7 -57,10 +46,8 @@@ import org.apache.rocketmq.streams.comm
import org.apache.rocketmq.streams.common.utils.MapKeyUtil;
import org.apache.rocketmq.streams.common.utils.StringUtil;
import org.apache.rocketmq.streams.common.utils.TraceUtil;
-import org.apache.rocketmq.streams.db.driver.orm.ORMUtil;
import org.apache.rocketmq.streams.window.debug.DebugWriter;
+ import org.apache.rocketmq.streams.window.minibatch.MiniBatchMsgCache;
import org.apache.rocketmq.streams.window.model.WindowCache;
import org.apache.rocketmq.streams.window.model.WindowInstance;
import org.apache.rocketmq.streams.window.operator.AbstractShuffleWindow;
@@@ -77,11 -90,17 +85,13 @@@ public class ShuffleChannel extends Abs
protected ShuffleCache shuffleCache;
- protected Map<String, ISplit> queueMap = new ConcurrentHashMap<>();
- protected List<ISplit> queueList;//所有的分片
+ protected Map<String, ISplit<?, ?>> queueMap = new ConcurrentHashMap<>();
+ /**
+ * 所有的分片
+ */
+ protected List<ISplit<?, ?>> queueList;
- // protected NotifyChannel notfiyChannel;//负责做分片的通知管理
protected AbstractShuffleWindow window;
- /**
- * 当前管理的分片
- */
- private Set<String> currentQueueIds;
protected transient boolean isWindowTest = false;
@@@ -116,28 -126,16 +117,16 @@@
* init shuffle channel
*/
public void init() {
- this.consumer = createSource(window.getNameSpace(), window.getConfigureName());
-
- this.producer = createSink(window.getNameSpace(), window.getConfigureName());
- if (this.consumer == null || this.producer == null) {
- autoCreateShuffleChannel(window.getFireReceiver().getPipeline());
- }
- if (this.consumer == null) {
- return;
- }
- if (this.consumer instanceof AbstractSource) {
- ((AbstractSource) this.consumer).setJsonData(true);
- }
+ init(this.window);
if (producer != null && (queueList == null || queueList.size() == 0)) {
queueList = producer.getSplitList();
- Map<String, ISplit> tmp = new ConcurrentHashMap<>();
- for (ISplit queue : queueList) {
+ Map<String, ISplit<?, ?>> tmp = new ConcurrentHashMap<>();
+ for (ISplit<?, ?> queue : queueList) {
tmp.put(queue.getQueueId(), queue);
}
-
this.queueMap = tmp;
}
- isWindowTest = ComponentCreator.getPropertyBooleanValue("window.fire.isTest");
+// isWindowTest = ComponentCreator.getPropertyBooleanValue("window.fire.isTest");
}
/**
@@@ -467,8 -505,11 +455,8 @@@
return splitNum > 0 ? splitNum : 32;
}
- public Set<String> getCurrentQueueIds() {
- return currentQueueIds;
- }
- public List<ISplit> getQueueList() {
+ public List<ISplit<?, ?>> getQueueList() {
return queueList;
}
diff --cc rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/trigger/WindowTrigger.java
index b012b175,69b6494a..e3bb3f83
--- a/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/trigger/WindowTrigger.java
+++ b/rocketmq-streams-window/src/main/java/org/apache/rocketmq/streams/window/trigger/WindowTrigger.java
@@@ -287,7 -287,7 +287,6 @@@ public class WindowTrigger extends Abst
}
});
for (WindowInstance windowInstance : windowInstanceList) {
- System.out.println("fire by finish flag");
- // System.out.println("fire by finish flag");
fireWindowInstance(windowInstance);
}
}