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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
