You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@eventmesh.apache.org by "Copilot (via GitHub)" <gi...@apache.org> on 2026/10/20 02:43:35 UTC
Re: [PR] [ISSUE #5384] Fix gRPC stream message loss during subscriber churn (eventmesh)
Copilot commented on code in PR #5429:
URL: https://github.com/apache/eventmesh/pull/5429#discussion_r4320942774
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/AbstractPushRequest.java:
##########
@@ -127,8 +140,13 @@ private void finish() {
}
protected void complete() {
+ if (complete.compareAndSet(Boolean.FALSE, Boolean.TRUE)) {
+ finish();
+ }
+ }
+
+ protected void completeWithoutOffsetUpdate() {
complete.compareAndSet(Boolean.FALSE, Boolean.TRUE);
- finish();
}
Review Comment:
`completeWithoutOffsetUpdate()` no longer calls `finish()`. This is a behavioral change from the previous implementation and can leave request bookkeeping (e.g., waiting-map cleanup / completion side-effects) unexecuted, potentially causing leaks or stuck requests. Consider mirroring `complete()` by calling `finish()` when transitioning from incomplete to complete (but keep offset-update avoidance within `finish()` or by splitting `finish()` responsibilities).
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/StreamPushRequest.java:
##########
@@ -29,98 +29,268 @@
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.RandomUtils;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
-
-import io.grpc.stub.StreamObserver;
+import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class StreamPushRequest extends AbstractPushRequest {
- private final Map<String, List<EventEmitter<CloudEvent>>> idcEmitters;
-
- private final List<EventEmitter<CloudEvent>> totalEmitters;
+ private final StreamTopicConfig topicConfig;
private final SubscriptionMode subscriptionMode;
private final int startIdx;
+ private final List<BroadcastTarget> pendingBroadcastTargets = new ArrayList<>();
+
+ private boolean broadcastInitialized;
+
+ private boolean nackPending;
+
public StreamPushRequest(HandleMsgContext handleMsgContext, Map<String, Set<AbstractPushRequest>> waitingRequests) {
super(handleMsgContext, waitingRequests);
- StreamTopicConfig topicConfig = (StreamTopicConfig) handleMsgContext.getConsumeTopicConfig();
- this.idcEmitters = topicConfig.getIdcEmitters();
- this.totalEmitters = topicConfig.getTotalEmitters();
+ this.topicConfig = (StreamTopicConfig) handleMsgContext.getConsumeTopicConfig();
this.subscriptionMode = topicConfig.getSubscriptionMode();
- this.startIdx = RandomUtils.nextInt(0, totalEmitters.size());
+ this.startIdx = RandomUtils.nextInt(0, Integer.MAX_VALUE);
}
@Override
- public void tryPushRequest() {
+ public synchronized void tryPushRequest() {
+ if (isComplete()) {
+ return;
+ }
+ if (nackPending) {
+ tryNack();
+ return;
+ }
if (eventMeshCloudEvent == null) {
+ if (supportsAsyncNack()) {
+ onRetriesExhausted();
+ }
+ return;
+ }
+
+ if (subscriptionMode == SubscriptionMode.BROADCASTING) {
+ tryPushBroadcast();
return;
}
List<EventEmitter<CloudEvent>> eventEmitters = selectEmitter();
+ if (eventEmitters.isEmpty()) {
+ if (supportsAsyncNack()) {
+ delayRetry();
+ }
+ return;
+ }
+
+ boolean retryNeeded = false;
for (EventEmitter<CloudEvent> eventEmitter : eventEmitters) {
- this.lastPushTime = System.currentTimeMillis();
+ if (eventEmitter.isClosed() || !trySend(eventEmitter)) {
+ retryNeeded = true;
+ }
+ }
+ if (retryNeeded) {
+ delayRetry();
+ } else {
+ complete();
+ }
+ }
- eventMeshCloudEvent = CloudEvent.newBuilder(eventMeshCloudEvent)
- .putAttributes(EventMeshConstants.REQ_EVENTMESH2C_TIMESTAMP,
- CloudEventAttributeValue.newBuilder().setCeString(String.valueOf(lastPushTime)).build())
- .build();
- try {
- // catch the error and retry, don't use eventEmitter.onNext() to hide the error
- StreamObserver<CloudEvent> emitter = eventEmitter.getEmitter();
- synchronized (emitter) {
- emitter.onNext(eventMeshCloudEvent);
- }
+ private void tryPushBroadcast() {
+ boolean firstAttempt = !broadcastInitialized;
+ if (!broadcastInitialized) {
+ List<EventEmitter<CloudEvent>> selected = selectEmitter();
+ if (selected.isEmpty()) {
+ return;
+ }
Review Comment:
In broadcast mode, first-attempt selection returning an empty list exits without scheduling a retry or completing. This can leave the request permanently pending (no timer is set) if there are temporarily no emitters or selection races with churn. Consider calling `delayRetry()` (or, if `topicConfig.getSize() == 0`, completing without offset update / recovery) rather than returning silently.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java:
##########
@@ -117,6 +126,21 @@ public List<EventEmitter<CloudEvent>> getTotalEmitters() {
return totalEmitters;
}
+ public synchronized String getEmitterKey(EventEmitter<CloudEvent> emitter) {
+ return emitterKeys.get(emitter);
+ }
+
+ public synchronized EventEmitter<CloudEvent> getEmitterByKey(String key) {
+ for (Map.Entry<String, Map<String, EventEmitter<CloudEvent>>> idc : idcEmitterMap.entrySet()) {
+ for (Map.Entry<String, EventEmitter<CloudEvent>> client : idc.getValue().entrySet()) {
+ if ((idc.getKey() + '\0' + client.getKey()).equals(key)) {
+ return client.getValue();
+ }
+ }
+ }
+ return null;
+ }
Review Comment:
`getEmitterByKey()` performs an O(N) scan over all registered emitters each lookup. In the new broadcast retry path, this lookup may happen repeatedly per message retry and can become a hotspot with many subscribers. Consider maintaining a direct key→emitter map updated in `registerClient`/`deregisterClient` to make lookups O(1).
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/consumer/consumergroup/StreamTopicConfig.java:
##########
@@ -117,6 +126,21 @@ public List<EventEmitter<CloudEvent>> getTotalEmitters() {
return totalEmitters;
}
Review Comment:
`getTotalEmitters()` returns the internal (mutable) `List` instance. Since this list is used in push selection and may be accessed concurrently, callers mutating it would corrupt internal state and can introduce hard-to-debug concurrency issues. A safer approach is to return an unmodifiable view or a defensive copy (and similarly for other emitter collection getters).
##########
eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java:
##########
@@ -378,6 +387,19 @@ private void submitConsumeRequestLater(final ConsumeRequest consumeRequest) {
}, 1000, TimeUnit.MILLISECONDS);
}
+ private void submitConsumeRequestLaterIndividually(
+ final List<MessageExt> msgs,
+ final ProcessQueue processQueue,
+ final MessageQueue messageQueue) {
+ this.scheduledExecutorService.schedule(() -> {
+ for (MessageExt msg : msgs) {
+ List<MessageExt> singleMessage = new ArrayList<>(1);
+ singleMessage.add(msg);
+ submitConsumeRequest(singleMessage, processQueue, messageQueue, true);
+ }
+ }, 5000, TimeUnit.MILLISECONDS);
Review Comment:
The retry delay `5000` is a new magic number distinct from the existing retry scheduling in this class. If this delay is intentional (e.g., aligning with RocketMQ retry semantics), consider extracting it into a named constant (or reusing an existing configurable delay) to document the rationale and make future tuning safer.
##########
eventmesh-storage-plugin/eventmesh-storage-rocketmq/src/main/java/org/apache/eventmesh/storage/rocketmq/consumer/PushConsumerImpl.java:
##########
@@ -102,6 +107,9 @@ public Properties attributes() {
}
public void start() {
+ if (rocketmqPushConsumer.getConsumeMessageBatchMaxSize() != 1) {
+ throw new StorageRuntimeException("EventMesh RocketMQ push consumer requires a consume message batch size of one");
+ }
Review Comment:
This exception message would be more actionable if it included the actual configured batch size value (e.g., `... got X`). That reduces time-to-diagnose when configuration/mocking overrides the constructor default.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/service/EventEmitter.java:
##########
@@ -25,20 +27,40 @@
public class EventEmitter<T> {
private final StreamObserver<T> emitter;
+ private final AtomicBoolean closed = new AtomicBoolean();
public EventEmitter(StreamObserver<T> emitter) {
this.emitter = emitter;
}
public synchronized void onNext(T event) {
+ if (closed.get()) {
+ return;
+ }
try {
emitter.onNext(event);
} catch (Exception e) {
+ closed.set(true);
log.warn("StreamObserver Error onNext. {}", e.getMessage());
}
Review Comment:
The log line drops the exception stack trace, which makes diagnosing stream failures harder (especially since this path also marks the stream closed). Consider logging the throwable as the last parameter so the stack trace is captured.
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/push/StreamPushRequest.java:
##########
@@ -29,98 +29,268 @@
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.RandomUtils;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
-
-import io.grpc.stub.StreamObserver;
+import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class StreamPushRequest extends AbstractPushRequest {
- private final Map<String, List<EventEmitter<CloudEvent>>> idcEmitters;
-
- private final List<EventEmitter<CloudEvent>> totalEmitters;
+ private final StreamTopicConfig topicConfig;
private final SubscriptionMode subscriptionMode;
private final int startIdx;
+ private final List<BroadcastTarget> pendingBroadcastTargets = new ArrayList<>();
+
+ private boolean broadcastInitialized;
+
+ private boolean nackPending;
+
public StreamPushRequest(HandleMsgContext handleMsgContext, Map<String, Set<AbstractPushRequest>> waitingRequests) {
super(handleMsgContext, waitingRequests);
- StreamTopicConfig topicConfig = (StreamTopicConfig) handleMsgContext.getConsumeTopicConfig();
- this.idcEmitters = topicConfig.getIdcEmitters();
- this.totalEmitters = topicConfig.getTotalEmitters();
+ this.topicConfig = (StreamTopicConfig) handleMsgContext.getConsumeTopicConfig();
this.subscriptionMode = topicConfig.getSubscriptionMode();
- this.startIdx = RandomUtils.nextInt(0, totalEmitters.size());
+ this.startIdx = RandomUtils.nextInt(0, Integer.MAX_VALUE);
}
@Override
- public void tryPushRequest() {
+ public synchronized void tryPushRequest() {
+ if (isComplete()) {
+ return;
+ }
+ if (nackPending) {
+ tryNack();
+ return;
+ }
if (eventMeshCloudEvent == null) {
+ if (supportsAsyncNack()) {
+ onRetriesExhausted();
+ }
+ return;
+ }
+
+ if (subscriptionMode == SubscriptionMode.BROADCASTING) {
+ tryPushBroadcast();
return;
}
List<EventEmitter<CloudEvent>> eventEmitters = selectEmitter();
+ if (eventEmitters.isEmpty()) {
+ if (supportsAsyncNack()) {
+ delayRetry();
+ }
+ return;
+ }
Review Comment:
When `selectEmitter()` returns empty and `supportsAsyncNack()` is false, the method returns without calling `delayRetry()` or completing the request. That can leave the push request neither rescheduled nor finished, effectively stalling delivery. A concrete fix is to ensure this branch either schedules a retry (for both modes) or completes (e.g., for cases where there are definitively no subscribers).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org