You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2023/05/06 08:35:17 UTC
[eventmesh] branch master updated: [ISSUE #3846]Remove the redundant method declaration in hierarchy. (#3847)
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 3686443ae [ISSUE #3846]Remove the redundant method declaration in hierarchy. (#3847)
3686443ae is described below
commit 3686443ae4a79cc7b0c12cfbee592588bf66f9ca
Author: pandaapo <35...@users.noreply.github.com>
AuthorDate: Sat May 6 16:35:12 2023 +0800
[ISSUE #3846]Remove the redundant method declaration in hierarchy. (#3847)
* Remove the redundant method declaration and unused field.
* Modify as review.
* Modify access modifier.
---
.../eventmesh/api/EventMeshAsyncConsumeContext.java | 14 ++++----------
.../storage/kafka/consumer/KafkaConsumerRunner.java | 8 +-------
.../storage/standalone/broker/task/Subscribe.java | 16 ++++++----------
3 files changed, 11 insertions(+), 27 deletions(-)
diff --git a/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/EventMeshAsyncConsumeContext.java b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/EventMeshAsyncConsumeContext.java
index 633a5b712..54a9c3f17 100644
--- a/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/EventMeshAsyncConsumeContext.java
+++ b/eventmesh-storage-plugin/eventmesh-storage-api/src/main/java/org/apache/eventmesh/api/EventMeshAsyncConsumeContext.java
@@ -17,19 +17,13 @@
package org.apache.eventmesh.api;
+import lombok.Getter;
+import lombok.Setter;
+@Getter
+@Setter
public abstract class EventMeshAsyncConsumeContext extends AsyncConsumeContext {
private AbstractContext abstractContext;
- public AbstractContext getAbstractContext() {
- return abstractContext;
- }
-
- public void setAbstractContext(AbstractContext abstractContext) {
- this.abstractContext = abstractContext;
- }
-
- public abstract void commit(EventMeshAction action);
-
}
\ No newline at end of file
diff --git a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerRunner.java b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerRunner.java
index 2ba0e6d34..e44075a32 100644
--- a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerRunner.java
+++ b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerRunner.java
@@ -28,12 +28,9 @@ import org.apache.kafka.common.errors.WakeupException;
import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
import io.cloudevents.CloudEvent;
-
import lombok.extern.slf4j.Slf4j;
@Slf4j
@@ -43,7 +40,6 @@ public class KafkaConsumerRunner implements Runnable {
private final KafkaConsumer<String, CloudEvent> consumer;
private ConcurrentHashMap<CloudEvent, Long> cloudEventToOffset;
private EventListener listener;
- private AtomicInteger offset;
public KafkaConsumerRunner(KafkaConsumer<String, CloudEvent> kafkaConsumer) {
this.consumer = kafkaConsumer;
@@ -81,13 +77,11 @@ public class KafkaConsumerRunner implements Runnable {
log.info("message commit, topic: {}, current offset:{}", topicName,
rec.offset());
break;
- case ReconsumeLater:
- // don't update offset
- break;
case ManualAck:
// update offset
log.info("message ack, topic: {}, current offset:{}", topicName, rec.offset());
break;
+ case ReconsumeLater:
default:
}
}
diff --git a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/task/Subscribe.java b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/task/Subscribe.java
index 2af625913..09e90259d 100644
--- a/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/task/Subscribe.java
+++ b/eventmesh-storage-plugin/eventmesh-storage-standalone/src/main/java/org/apache/eventmesh/storage/standalone/broker/task/Subscribe.java
@@ -33,9 +33,9 @@ import lombok.extern.slf4j.Slf4j;
public class Subscribe {
@Getter
- private String topicName;
- private StandaloneBroker standaloneBroker;
- private EventListener listener;
+ private final String topicName;
+ private final StandaloneBroker standaloneBroker;
+ private final EventListener listener;
@Getter
private volatile boolean isRunning;
@Getter
@@ -73,18 +73,14 @@ public class Subscribe {
switch (action) {
case CommitMessage:
// update offset
- log.info("message commit, topic: {}, current offset:{}", topicName,
- offset.get());
- break;
- case ReconsumeLater:
- // don't update offset
+ log.info("message commit, topic: {}, current offset:{}", topicName, offset.get());
break;
case ManualAck:
// update offset
offset.incrementAndGet();
- log
- .info("message ack, topic: {}, current offset:{}", topicName, offset.get());
+ log.info("message ack, topic: {}, current offset:{}", topicName, offset.get());
break;
+ case ReconsumeLater:
default:
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org