This is an automated email from the ASF dual-hosted git repository.
mxsm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 50d6113ce [ISSUE #4062]Implement the method
printRetryThreadPoolState() in EventMeshTcpRetryer (#4063)
50d6113ce is described below
commit 50d6113cec1cdbbb38825625c5b89dffa62f604f
Author: pandaapo <[email protected]>
AuthorDate: Mon Aug 21 09:12:51 2023 +0800
[ISSUE #4062]Implement the method printRetryThreadPoolState() in
EventMeshTcpRetryer (#4063)
* Complete the method printRetryThreadPoolState().
* Modify as review.
* Modify as review.
* Implement the new method of Connector in mongodb conector. Optimize
duplicate code in redis connector.
* Implement the new method of Connector in S3 connector module. Optimize
some duplicate codes.
* Fix checkstyle error.
* Fix checkstyle error.
* Fix checkstyle error.
* Fix checkstyle error.
---
.../kafka/sink/connector/KafkaSinkConnector.java | 20 ++----
.../source/connector/KafkaSourceConnector.java | 18 ++----
.../sink/connector/MongodbSinkConnector.java | 13 ++++
.../source/connector/MongodbSourceConnector.java | 13 ++++
.../sink/connector/OpenFunctionSinkConnector.java | 2 +-
.../connector/OpenFunctionSourceConnector.java | 2 +-
.../pulsar/sink/connector/PulsarSinkConnector.java | 13 ++--
.../source/connector/PulsarSourceConnector.java | 14 ++---
.../redis/sink/connector/RedisSinkConnector.java | 11 ++--
.../source/connector/RedisSourceConnector.java | 12 ++--
.../source/connector/RocketMQSourceConnector.java | 11 +---
.../s3/source/connector/S3SourceConnector.java | 13 ++++
.../apache/eventmesh/openconnect/Application.java | 2 -
.../apache/eventmesh/openconnect/SinkWorker.java | 2 +-
.../api/connector/ConnectorContext.java | 3 -
.../api/connector/SinkConnectorContext.java | 2 -
.../api/connector/SourceConnectorContext.java | 1 -
.../offsetmgmt/api/data/RecordOffset.java | 2 +-
.../api/storage/ConnectorRecordPartition.java | 1 -
.../offsetmgmt/nacos/NacosConfigService.java | 6 +-
.../client/session/retry/EventMeshTcpRetryer.java | 2 +
.../eventmesh/runtime/util/ThreadPoolHelper.java | 46 ++++++++++++++
.../runtime/util/ThreadPoolHelperTest.java | 71 ++++++++++++++++++++++
23 files changed, 202 insertions(+), 78 deletions(-)
diff --git
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/connector/KafkaSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/connector/KafkaSinkConnector.java
index 620a488f1..49b7e88f1 100644
---
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/connector/KafkaSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/sink/connector/KafkaSinkConnector.java
@@ -54,25 +54,17 @@ public class KafkaSinkConnector implements Sink {
@Override
public void init(Config config) {
this.sinkConfig = (KafkaSinkConfig) config;
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
sinkConfig.getConnectorConfig().getBootstrapServers());
- props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
sinkConfig.getConnectorConfig().getKeyConverter());
- props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
sinkConfig.getConnectorConfig().getValueConverter());
- props.put(ProducerConfig.ACKS_CONFIG,
sinkConfig.getConnectorConfig().getAck());
- props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,
sinkConfig.getConnectorConfig().getMaxRequestSize());
- props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,
sinkConfig.getConnectorConfig().getBufferMemory());
- props.put(ProducerConfig.BATCH_SIZE_CONFIG,
sinkConfig.getConnectorConfig().getBatchSize());
- props.put(ProducerConfig.LINGER_MS_CONFIG,
sinkConfig.getConnectorConfig().getLingerMs());
- props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,
sinkConfig.getConnectorConfig().getRequestTimeoutMs());
- props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,
sinkConfig.getConnectorConfig().getMaxInFightRequestsPerConnection());
- props.put(ProducerConfig.RETRIES_CONFIG,
sinkConfig.getConnectorConfig().getRetries());
- props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
sinkConfig.getConnectorConfig().getCompressionType());
- producer = new KafkaProducer<>(props);
+ doInit();
}
@Override
public void init(ConnectorContext connectorContext) throws Exception {
- SinkConnectorContext sinkConnectorContext =
(SinkConnectorContext)connectorContext;
+ SinkConnectorContext sinkConnectorContext = (SinkConnectorContext)
connectorContext;
this.sinkConfig = (KafkaSinkConfig)
sinkConnectorContext.getSinkConfig();
+ doInit();
+ }
+
+ private void doInit() {
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
sinkConfig.getConnectorConfig().getBootstrapServers());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
sinkConfig.getConnectorConfig().getKeyConverter());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
sinkConfig.getConnectorConfig().getValueConverter());
diff --git
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
index 2a7c940d0..18fa42163 100644
---
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
@@ -55,23 +55,17 @@ public class KafkaSourceConnector implements Source {
@Override
public void init(Config config) throws Exception {
this.sourceConfig = (KafkaSourceConfig) config;
- Properties props = new Properties();
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
sourceConfig.getConnectorConfig().getBootstrapServers());
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
sourceConfig.getConnectorConfig().getKeyConverter());
- props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
sourceConfig.getConnectorConfig().getValueConverter());
- props.put(ConsumerConfig.GROUP_ID_CONFIG,
sourceConfig.getConnectorConfig().getGroupID());
- props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
sourceConfig.getConnectorConfig().getEnableAutoCommit());
- props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,
sourceConfig.getConnectorConfig().getMaxPollRecords());
- props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,
sourceConfig.getConnectorConfig().getAutoCommitIntervalMS());
- props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,
sourceConfig.getConnectorConfig().getSessionTimeoutMS());
- this.pollTimeOut = sourceConfig.getConnectorConfig().getPollTimeOut();
- this.kafkaConsumer = new KafkaConsumer<String, String>(props);
+ doInit();
}
@Override
public void init(ConnectorContext connectorContext) throws Exception {
- SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext)connectorContext;
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
this.sourceConfig = (KafkaSourceConfig)
sourceConnectorContext.getSourceConfig();
+ doInit();
+ }
+
+ private void doInit() {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
sourceConfig.getConnectorConfig().getBootstrapServers());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
sourceConfig.getConnectorConfig().getKeyConverter());
diff --git
a/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/sink/connector/MongodbSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/sink/connector/MongodbSinkConnector.java
index aa11cb87a..ef65ff6e9 100644
---
a/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/sink/connector/MongodbSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/sink/connector/MongodbSinkConnector.java
@@ -22,6 +22,8 @@ import
org.apache.eventmesh.connector.mongodb.sink.client.MongodbReplicaSetSinkC
import
org.apache.eventmesh.connector.mongodb.sink.client.MongodbStandaloneSinkClient;
import org.apache.eventmesh.connector.mongodb.sink.config.MongodbSinkConfig;
import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext;
import org.apache.eventmesh.openconnect.api.sink.Sink;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
@@ -51,6 +53,17 @@ public class MongodbSinkConnector implements Sink {
@Override
public void init(Config config) throws Exception {
this.sinkConfig = (MongodbSinkConfig) config;
+ doInit();
+ }
+
+ @Override
+ public void init(ConnectorContext connectorContext) throws Exception {
+ SinkConnectorContext sinkConnectorContext = (SinkConnectorContext)
connectorContext;
+ this.sinkConfig = (MongodbSinkConfig)
sinkConnectorContext.getSinkConfig();
+ doInit();
+ }
+
+ private void doInit() {
String connectorType =
sinkConfig.getConnectorConfig().getConnectorType();
if (connectorType.equals(ClusterType.STANDALONE.name())) {
this.client = new
MongodbStandaloneSinkClient(sinkConfig.getConnectorConfig());
diff --git
a/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/source/connector/MongodbSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/source/connector/MongodbSourceConnector.java
index baaa58a0d..995f8cd4a 100644
---
a/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/source/connector/MongodbSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-mongodb/src/main/java/org/apache/eventmesh/connector/mongodb/source/connector/MongodbSourceConnector.java
@@ -22,6 +22,8 @@ import
org.apache.eventmesh.connector.mongodb.source.client.MongodbReplicaSetSou
import
org.apache.eventmesh.connector.mongodb.source.client.MongodbStandaloneSourceClient;
import
org.apache.eventmesh.connector.mongodb.source.config.MongodbSourceConfig;
import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SourceConnectorContext;
import org.apache.eventmesh.openconnect.api.source.Source;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
@@ -55,6 +57,17 @@ public class MongodbSourceConnector implements Source {
@Override
public void init(Config config) throws Exception {
this.sourceConfig = (MongodbSourceConfig) config;
+ doInit();
+ }
+
+ @Override
+ public void init(ConnectorContext connectorContext) throws Exception {
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
+ this.sourceConfig = (MongodbSourceConfig)
sourceConnectorContext.getSourceConfig();
+ doInit();
+ }
+
+ private void doInit() {
this.queue = new LinkedBlockingQueue<>(1000);
String connectorType =
sourceConfig.getConnectorConfig().getConnectorType();
if (connectorType.equals(ClusterType.STANDALONE.name())) {
diff --git
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
index b3f488456..aed693600 100644
---
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/sink/connector/OpenFunctionSinkConnector.java
@@ -54,7 +54,7 @@ public class OpenFunctionSinkConnector implements Sink {
@Override
public void init(ConnectorContext connectorContext) throws Exception {
// init config for openfunction source connector
- SinkConnectorContext sinkConnectorContext =
(SinkConnectorContext)connectorContext;
+ SinkConnectorContext sinkConnectorContext = (SinkConnectorContext)
connectorContext;
this.sinkConfig = (OpenFunctionSinkConfig)
sinkConnectorContext.getSinkConfig();
this.queue = new LinkedBlockingQueue<>(1000);
}
diff --git
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
index 65877cfde..916d6793d 100644
---
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
@@ -55,7 +55,7 @@ public class OpenFunctionSourceConnector implements Source {
@Override
public void init(ConnectorContext connectorContext) throws Exception {
- SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext)connectorContext;
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
// init config for openfunction source connector
this.sourceConfig = (OpenFunctionSourceConfig)
sourceConnectorContext.getSourceConfig();
this.queue = new LinkedBlockingQueue<>(1000);
diff --git
a/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/sink/connector/PulsarSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/sink/connector/PulsarSinkConnector.java
index 06c882dc8..426211cdd 100644
---
a/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/sink/connector/PulsarSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/sink/connector/PulsarSinkConnector.java
@@ -53,19 +53,18 @@ public class PulsarSinkConnector implements Sink {
public void init(Config config) throws Exception {
// init config for pulsar source connector
this.sinkConfig = (PulsarSinkConfig) config;
- PulsarClient client = PulsarClient.builder()
- .serviceUrl(sinkConfig.getConnectorConfig().getServiceUrl())
- .build();
- producer = client.newProducer()
- .topic(sinkConfig.getConnectorConfig().getTopic())
- .create();
+ doInit();
}
@Override
public void init(ConnectorContext connectorContext) throws Exception {
// init config for pulsar source connector
- SinkConnectorContext sinkConnectorContext =
(SinkConnectorContext)connectorContext;
+ SinkConnectorContext sinkConnectorContext = (SinkConnectorContext)
connectorContext;
this.sinkConfig = (PulsarSinkConfig)
sinkConnectorContext.getSinkConfig();
+ doInit();
+ }
+
+ private void doInit() throws Exception {
PulsarClient client = PulsarClient.builder()
.serviceUrl(sinkConfig.getConnectorConfig().getServiceUrl())
.build();
diff --git
a/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/source/connector/PulsarSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/source/connector/PulsarSourceConnector.java
index 5a5736bc6..9d77c5b6e 100644
---
a/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/source/connector/PulsarSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-pulsar/src/main/java/org/apache/eventmesh/connector/pulsar/source/connector/PulsarSourceConnector.java
@@ -59,19 +59,17 @@ public class PulsarSourceConnector implements Source {
public void init(Config config) throws Exception {
// init config for rocketmq source connector
this.sourceConfig = (PulsarSourceConfig) config;
- PulsarClient client = PulsarClient.builder()
- .serviceUrl(sourceConfig.getConnectorConfig().getServiceUrl())
- .build();
- consumer = client.newConsumer()
- .topic(sourceConfig.connectorConfig.getTopic())
- .subscriptionName(sourceConfig.getPubSubConfig().getGroup())
- .subscribe();
+ doInit();
}
@Override
public void init(ConnectorContext connectorContext) throws Exception {
- SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext)connectorContext;
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
this.sourceConfig = (PulsarSourceConfig)
sourceConnectorContext.getSourceConfig();
+ doInit();
+ }
+
+ private void doInit() throws Exception {
PulsarClient client = PulsarClient.builder()
.serviceUrl(sourceConfig.getConnectorConfig().getServiceUrl())
.build();
diff --git
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
index abc11b9c3..88246cc05 100644
---
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
@@ -55,16 +55,17 @@ public class RedisSinkConnector implements Sink {
@Override
public void init(Config config) throws Exception {
this.sinkConfig = (RedisSinkConfig) config;
- org.redisson.config.Config redisConfig = new
org.redisson.config.Config();
-
redisConfig.useSingleServer().setAddress(sinkConfig.connectorConfig.getServer());
- redisConfig.setCodec(CloudEventCodec.getInstance());
- this.redissonClient = Redisson.create(redisConfig);
+ doInit();
}
@Override
public void init(ConnectorContext connectorContext) throws Exception {
- SinkConnectorContext sinkConnectorContext =
(SinkConnectorContext)connectorContext;
+ SinkConnectorContext sinkConnectorContext = (SinkConnectorContext)
connectorContext;
this.sinkConfig = (RedisSinkConfig)
sinkConnectorContext.getSinkConfig();
+ doInit();
+ }
+
+ private void doInit() {
org.redisson.config.Config redisConfig = new
org.redisson.config.Config();
redisConfig.useSingleServer().setAddress(sinkConfig.connectorConfig.getServer());
redisConfig.setCodec(CloudEventCodec.getInstance());
diff --git
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
index 985d4c3b6..f42f7a3aa 100644
---
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
@@ -58,17 +58,17 @@ public class RedisSourceConnector implements Source {
@Override
public void init(Config config) throws Exception {
this.sourceConfig = (RedisSourceConfig) config;
- org.redisson.config.Config redisConfig = new
org.redisson.config.Config();
-
redisConfig.useSingleServer().setAddress(sourceConfig.connectorConfig.getServer());
- redisConfig.setCodec(CloudEventCodec.getInstance());
- this.redissonClient = Redisson.create(redisConfig);
- this.queue = new LinkedBlockingQueue<>(1000);
+ doInit();
}
@Override
public void init(ConnectorContext connectorContext) throws Exception {
- SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext)connectorContext;
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
this.sourceConfig = (RedisSourceConfig)
sourceConnectorContext.getSourceConfig();
+ doInit();
+ }
+
+ private void doInit() {
org.redisson.config.Config redisConfig = new
org.redisson.config.Config();
redisConfig.useSingleServer().setAddress(sourceConfig.connectorConfig.getServer());
redisConfig.setCodec(CloudEventCodec.getInstance());
diff --git
a/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/source/connector/RocketMQSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/source/connector/RocketMQSourceConnector.java
index 9fc68b010..331641d22 100644
---
a/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/source/connector/RocketMQSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/source/connector/RocketMQSourceConnector.java
@@ -25,35 +25,26 @@ import org.apache.eventmesh.openconnect.api.source.Source;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
-import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.ConnectorRecordPartition;
import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetStorageReader;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.TreeMap;
-import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
import lombok.extern.slf4j.Slf4j;
@@ -208,7 +199,7 @@ public class RocketMQSourceConnector implements Source {
prepareCommitOffset.forEach((messageQueue, list) -> {
Iterator<AtomicLong> offsetIterator = list.iterator();
- while(offsetIterator.hasNext()) {
+ while (offsetIterator.hasNext()) {
Map<MessageQueue, Long> commitOffsetTable = new HashMap<>();
commitOffsetTable.put(messageQueue,
offsetIterator.next().get());
consumer.commitSync(commitOffsetTable, false);
diff --git
a/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/connector/S3SourceConnector.java
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/connector/S3SourceConnector.java
index 43b757d08..8cd0a3627 100644
---
a/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/connector/S3SourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/connector/S3SourceConnector.java
@@ -21,6 +21,8 @@ package org.apache.eventmesh.connector.s3.source.connector;
import org.apache.eventmesh.connector.s3.source.config.S3SourceConfig;
import org.apache.eventmesh.connector.s3.source.config.SourceConnectorConfig;
import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SourceConnectorContext;
import org.apache.eventmesh.openconnect.api.source.Source;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
@@ -79,6 +81,17 @@ public class S3SourceConnector implements Source {
public void init(Config config) throws Exception {
// init config for s3 source connector
this.sourceConfig = (S3SourceConfig) config;
+ doInit();
+ }
+
+ @Override
+ public void init(ConnectorContext connectorContext) throws Exception {
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
+ this.sourceConfig = (S3SourceConfig)
sourceConnectorContext.getSourceConfig();
+ doInit();
+ }
+
+ private void doInit() {
this.sourceConnectorConfig =
this.sourceConfig.getSourceConnectorConfig();
this.eachRecordSize = calculateEachRecordSize();
AwsBasicCredentials basicCredentials =
AwsBasicCredentials.create(this.sourceConnectorConfig.getAccessKey(),
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
index 25aaf2b6b..8a0341f41 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/Application.java
@@ -21,8 +21,6 @@ import org.apache.eventmesh.openconnect.api.config.Config;
import org.apache.eventmesh.openconnect.api.config.SinkConfig;
import org.apache.eventmesh.openconnect.api.config.SourceConfig;
import org.apache.eventmesh.openconnect.api.connector.Connector;
-import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
-import org.apache.eventmesh.openconnect.api.connector.SourceConnectorContext;
import org.apache.eventmesh.openconnect.api.sink.Sink;
import org.apache.eventmesh.openconnect.api.source.Source;
import org.apache.eventmesh.openconnect.util.ConfigUtil;
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SinkWorker.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SinkWorker.java
index 80718cde6..7ac918a3e 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SinkWorker.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SinkWorker.java
@@ -48,7 +48,7 @@ public class SinkWorker implements ConnectorWorker {
private final EventMeshTCPClient<CloudEvent> eventMeshTCPClient;
- public SinkWorker(Sink sink, SinkConfig config){
+ public SinkWorker(Sink sink, SinkConfig config) {
this.sink = sink;
this.config = config;
eventMeshTCPClient = buildEventMeshSubClient(config);
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/ConnectorContext.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/ConnectorContext.java
index a17e21126..bc4298fde 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/ConnectorContext.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/ConnectorContext.java
@@ -17,9 +17,6 @@
package org.apache.eventmesh.openconnect.api.connector;
-import org.apache.eventmesh.openconnect.api.config.Config;
-import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
-
/**
* Connector Context
*/
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SinkConnectorContext.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SinkConnectorContext.java
index 22b94dbcb..a7aef4fff 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SinkConnectorContext.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SinkConnectorContext.java
@@ -18,8 +18,6 @@
package org.apache.eventmesh.openconnect.api.connector;
import org.apache.eventmesh.openconnect.api.config.SinkConfig;
-import org.apache.eventmesh.openconnect.api.config.SourceConfig;
-import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetStorageReader;
import lombok.Data;
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SourceConnectorContext.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SourceConnectorContext.java
index 72af997c5..14fd92218 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SourceConnectorContext.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/connector/SourceConnectorContext.java
@@ -17,7 +17,6 @@
package org.apache.eventmesh.openconnect.api.connector;
-import org.apache.eventmesh.openconnect.api.config.Config;
import org.apache.eventmesh.openconnect.api.config.SourceConfig;
import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetStorageReader;
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/data/RecordOffset.java
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/data/RecordOffset.java
index cad0e62cb..fee70de27 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/data/RecordOffset.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/data/RecordOffset.java
@@ -29,7 +29,7 @@ public class RecordOffset {
*/
private Map<String, ?> offset = new HashMap<>();
- public RecordOffset(){
+ public RecordOffset() {
}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/storage/ConnectorRecordPartition.java
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/storage/ConnectorRecordPartition.java
index 56ffd7756..05308e800 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/storage/ConnectorRecordPartition.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-api/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/api/storage/ConnectorRecordPartition.java
@@ -19,7 +19,6 @@ package
org.apache.eventmesh.openconnect.offsetmgmt.api.storage;
import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
-import java.io.Serializable;
import java.util.Map;
import java.util.Objects;
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-nacos/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/nacos/NacosConfigService.java
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-nacos/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/nacos/NacosConfigService.java
index 384062289..c8266d615 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-nacos/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/nacos/NacosConfigService.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-offsetmgmt-plugin/eventmesh-openconnect-offsetmgmt-nacos/src/main/java/org/apache/eventmesh/openconnect/offsetmgmt/nacos/NacosConfigService.java
@@ -28,7 +28,6 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.Executor;
import com.alibaba.nacos.api.NacosFactory;
@@ -112,7 +111,7 @@ public class NacosConfigService implements
OffsetManagementService {
Map<ConnectorRecordPartition, RecordOffset> recordMap =
positionStore.getKVMap();
List<Map<String, Object>> recordToSyncList = new ArrayList<>();
- for(Map.Entry<ConnectorRecordPartition, RecordOffset> entry :
recordMap.entrySet()) {
+ for (Map.Entry<ConnectorRecordPartition, RecordOffset> entry :
recordMap.entrySet()) {
Map<String, Object> synchronizeMap = new HashMap<>();
synchronizeMap.put("connectorRecordPartition", entry.getKey());
synchronizeMap.put("recordOffset", entry.getValue());
@@ -206,7 +205,8 @@ public class NacosConfigService implements
OffsetManagementService {
});
for (Map<String, Object> recordPartitionOffsetMap :
recordOffsetList) {
- ConnectorRecordPartition connectorRecordPartition =
JacksonUtils.toObj(JacksonUtils.toJson(recordPartitionOffsetMap.get("connectorRecordPartition")),
+ ConnectorRecordPartition connectorRecordPartition =
JacksonUtils.toObj(
+
JacksonUtils.toJson(recordPartitionOffsetMap.get("connectorRecordPartition")),
ConnectorRecordPartition.class);
RecordOffset recordOffset =
JacksonUtils.toObj(JacksonUtils.toJson(recordPartitionOffsetMap.get("recordOffset")),
RecordOffset.class);
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
index cc57f8854..1975a0901 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.DownStreamMsgContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
+import org.apache.eventmesh.runtime.util.ThreadPoolHelper;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
@@ -127,5 +128,6 @@ public class EventMeshTcpRetryer {
}
public void printRetryThreadPoolState() {
+ ThreadPoolHelper.printState(pool);
}
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/ThreadPoolHelper.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/ThreadPoolHelper.java
new file mode 100644
index 000000000..7f5b4bb31
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/ThreadPoolHelper.java
@@ -0,0 +1,46 @@
+/*
+ * 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.eventmesh.runtime.util;
+
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class ThreadPoolHelper {
+
+ public static void printState(ThreadPoolExecutor threadPool) {
+ if (threadPool == null) {
+ log.warn("No thread pool is provided!");
+ return;
+ }
+
+ log.info("The ThreadPool's state==================");
+ log.info("Shutdown | Terminating | Terminated: {} | {} | {}",
threadPool.isShutdown(), threadPool.isTerminating(), threadPool.isTerminated());
+ log.info("Active Threads: " + threadPool.getActiveCount());
+ log.info("Completed Tasks / Tasks: {} / {}" +
threadPool.getCompletedTaskCount(), threadPool.getTaskCount());
+ log.info("Queue Size: " + threadPool.getQueue().size());
+ log.info("Core Pool Size: " + threadPool.getCorePoolSize());
+ log.info("Maximum Pool Size: " + threadPool.getMaximumPoolSize());
+ log.info("Keep Alive Time(ms): " +
threadPool.getKeepAliveTime(TimeUnit.MILLISECONDS));
+ log.info("The rejection policy: " +
threadPool.getRejectedExecutionHandler().getClass().getSimpleName());
+ log.info("========================================");
+ }
+
+}
diff --git
a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/util/ThreadPoolHelperTest.java
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/util/ThreadPoolHelperTest.java
new file mode 100644
index 000000000..f117ee01c
--- /dev/null
+++
b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/util/ThreadPoolHelperTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.eventmesh.runtime.util;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ThreadPoolHelperTest {
+
+ @Mock
+ private ThreadPoolExecutor mockThreadPool;
+
+ @Mock
+ private RejectedExecutionHandler mockRejectionPolicy;
+
+ @Test
+ public void testPrintState() {
+ when(mockThreadPool.isShutdown()).thenReturn(false);
+ when(mockThreadPool.isTerminating()).thenReturn(false);
+ when(mockThreadPool.isTerminated()).thenReturn(false);
+ when(mockThreadPool.getActiveCount()).thenReturn(2);
+ when(mockThreadPool.getCompletedTaskCount()).thenReturn(10L);
+ when(mockThreadPool.getTaskCount()).thenReturn(15L);
+ when(mockThreadPool.getQueue()).thenReturn(new
LinkedBlockingDeque<>(5));
+ when(mockThreadPool.getCorePoolSize()).thenReturn(5);
+ when(mockThreadPool.getMaximumPoolSize()).thenReturn(10);
+
when(mockThreadPool.getKeepAliveTime(any(TimeUnit.class))).thenReturn(1000L);
+
when(mockThreadPool.getRejectedExecutionHandler()).thenReturn(mockRejectionPolicy);
+
+ ThreadPoolHelper.printState(mockThreadPool);
+
+ verify(mockThreadPool).isShutdown();
+ verify(mockThreadPool).isTerminating();
+ verify(mockThreadPool).isTerminated();
+ verify(mockThreadPool).getActiveCount();
+ verify(mockThreadPool).getCompletedTaskCount();
+ verify(mockThreadPool).getTaskCount();
+ verify(mockThreadPool).getQueue();
+ verify(mockThreadPool).getCorePoolSize();
+ verify(mockThreadPool).getMaximumPoolSize();
+ verify(mockThreadPool).getKeepAliveTime(any(TimeUnit.class));
+ verify(mockThreadPool).getRejectedExecutionHandler();
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
