This is an automated email from the ASF dual-hosted git repository.
mikexue 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 fb9157005 [ISSUE #3724]Do some code optimization[PulsarConsumerImpl]
fb9157005 is described below
commit fb915700565dea1e4d2264dc05ed83f3c8a4d423
Author: pandaapo <[email protected]>
AuthorDate: Wed May 3 13:24:10 2023 +0800
[ISSUE #3724]Do some code optimization[PulsarConsumerImpl]
---
.../eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java | 4 ++--
.../storage/knative/consumer/KnativeConsumerImpl.java | 13 ++++---------
.../eventmesh/storage/pravega/PravegaConsumerImpl.java | 2 +-
.../storage/pulsar/consumer/PulsarConsumerImpl.java | 11 ++++++-----
.../standalone/consumer/StandaloneConsumerAdaptor.java | 6 +++---
5 files changed, 16 insertions(+), 20 deletions(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java
index 0c88ec934..0fa75d384 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java
@@ -41,7 +41,7 @@ public class KafkaConsumerImpl implements Consumer {
private ClientConfiguration clientConfiguration;
@Override
- public synchronized void init(Properties props) throws Exception {
+ public synchronized void init(Properties props) {
String namesrvAddr = clientConfiguration.getNamesrvAddr();
String consumerGroup = props.getProperty("consumerGroup");
@@ -57,7 +57,7 @@ public class KafkaConsumerImpl implements Consumer {
}
@Override
- public void subscribe(String topic) throws Exception {
+ public void subscribe(String topic) {
consumer.subscribe(topic);
}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java
index 4b820d95f..fa7171031 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-knative/src/main/java/org/apache/eventmesh/storage/knative/consumer/KnativeConsumerImpl.java
@@ -26,26 +26,21 @@ import
org.apache.eventmesh.storage.knative.config.ClientConfiguration;
import java.util.List;
import java.util.Properties;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import io.cloudevents.CloudEvent;
+import lombok.extern.slf4j.Slf4j;
+
@Config(field = "clientConfiguration")
+@Slf4j
public class KnativeConsumerImpl implements Consumer {
- private static final Logger LOG =
LoggerFactory.getLogger(KnativeConsumerImpl.class);
-
private transient PullConsumerImpl pullConsumer;
-
/**
* Unified configuration class corresponding to knative-client.properties
*/
private ClientConfiguration clientConfiguration;
- private static final Logger logger =
LoggerFactory.getLogger(KnativeConsumerImpl.class);
-
@Override
public synchronized void init(Properties properties) throws Exception {
// Load parameters from properties file:
@@ -65,7 +60,7 @@ public class KnativeConsumerImpl implements Consumer {
try {
pullConsumer.unsubscribe(topic);
} catch (Exception e) {
- LOG.error("unsubscribe error", e);
+ log.error("unsubscribe error", e);
}
}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
index 528be1302..2e8b2b693 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
@@ -85,7 +85,7 @@ public class PravegaConsumerImpl implements Consumer {
}
@Override
- public void subscribe(String topic) throws Exception {
+ public void subscribe(String topic) {
if (!client.subscribe(topic, isBroadcast, consumerGroup, instanceName,
eventListener)) {
throw new PravegaStorageException(String.format("subscribe
topic[%s] fail.", topic));
}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/consumer/PulsarConsumerImpl.java
b/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/consumer/PulsarConsumerImpl.java
index 9601dd8a4..4fedb613d 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/consumer/PulsarConsumerImpl.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-pulsar/src/main/java/org/apache/eventmesh/storage/pulsar/consumer/PulsarConsumerImpl.java
@@ -38,12 +38,14 @@ import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import java.util.List;
+import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.format.EventDeserializationException;
+import io.cloudevents.core.format.EventFormat;
import io.cloudevents.core.provider.EventFormatProvider;
import io.cloudevents.jackson.JsonFormat;
@@ -61,7 +63,7 @@ public class PulsarConsumerImpl implements Consumer {
private PulsarClient pulsarClient;
private EventListener eventListener;
- private ConcurrentHashMap<String,
org.apache.pulsar.client.api.Consumer<byte[]>> consumerMap = new
ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String,
org.apache.pulsar.client.api.Consumer<byte[]>> consumerMap = new
ConcurrentHashMap<>();
/**
* Unified configuration class corresponding to pulsar-client.properties
@@ -129,10 +131,9 @@ public class PulsarConsumerImpl implements Consumer {
.subscriptionType(type)
.messageListener(
(MessageListener<byte[]>) (ackConsumer, msg) -> {
- CloudEvent cloudEvent = EventFormatProvider
- .getInstance()
- .resolveFormat(JsonFormat.CONTENT_TYPE)
- .deserialize(msg.getData());
+ EventFormat eventFormat = Objects.requireNonNull(
+
EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE));
+ CloudEvent cloudEvent =
eventFormat.deserialize(msg.getData());
eventListener.consume(cloudEvent, consumeContext);
try {
ackConsumer.acknowledge(msg);
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/consumer/StandaloneConsumerAdaptor.java
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/consumer/StandaloneConsumerAdaptor.java
index fcd631fa3..328faeeff 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/consumer/StandaloneConsumerAdaptor.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/consumer/StandaloneConsumerAdaptor.java
@@ -26,13 +26,13 @@ import java.util.Properties;
import io.cloudevents.CloudEvent;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor
public class StandaloneConsumerAdaptor implements Consumer {
private StandaloneConsumer consumer;
- public StandaloneConsumerAdaptor() {
- }
-
@Override
public boolean isStarted() {
return consumer.isStarted();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
