setup

Author: VGalaxies

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/subscription/payload/poll/ErrorPayload.java

@@ -28,6 +28,10 @@
 
 public class ErrorPayload implements SubscriptionPollPayload {
 
+  private static final String OUTDATED_ERROR_MSG = "outdated subscription event";
+  public static final ErrorPayload OUTDATED_ERROR_PAYLOAD =
+      new ErrorPayload(OUTDATED_ERROR_MSG, false);
+
   /** The error message describing the issue. */
   private transient String errorMessage;
 

iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java

@@ -1022,6 +1022,10 @@ private Optional<SubscriptionMessage> pollTabletsInternal(
 
             final String errorMessage = ((ErrorPayload) payload).getErrorMessage();
             final boolean critical = ((ErrorPayload) payload).isCritical();
+            if (Objects.equals(payload, ErrorPayload.OUTDATED_ERROR_PAYLOAD)) {
+              // suppress warn log when poll outdated subscription event
+              return Optional.empty();
+            }
             LOGGER.warn(
                 "Error occurred when SubscriptionConsumer {} polling tablets with commit context {}: {}, critical: {}",
                 this,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java

@@ -116,6 +116,15 @@ public List<SubscriptionCommitContext> commit(
     return broker.commit(consumerId, commitContexts, nack);
   }
 
+  public boolean isCommitContextOutdated(final SubscriptionCommitContext commitContext) {
+    final String consumerGroupId = commitContext.getConsumerGroupId();
+    final SubscriptionBroker broker = consumerGroupIdToSubscriptionBroker.get(consumerGroupId);
+    if (Objects.isNull(broker)) {
+      return true;
+    }
+    return broker.isCommitContextOutdated(commitContext);
+  }
+
   /////////////////////////////// broker ///////////////////////////////
 
   public boolean isBrokerExist(final String consumerGroupId) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java

@@ -329,6 +329,16 @@ public List<SubscriptionCommitContext> commit(
     return successfulCommitContexts;
   }
 
+  public boolean isCommitContextOutdated(final SubscriptionCommitContext commitContext) {
+    final String topicName = commitContext.getTopicName();
+    final SubscriptionPrefetchingQueue prefetchingQueue =
+        topicNameToPrefetchingQueue.get(topicName);
+    if (Objects.isNull(prefetchingQueue)) {
+      return true;
+    }
+    return prefetchingQueue.isCommitContextOutdated(commitContext);
+  }
+
   /////////////////////////////// prefetching queue ///////////////////////////////
 
   public void bindPrefetchingQueue(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java

@@ -68,6 +68,8 @@ public abstract class SubscriptionPrefetchingQueue {
   private final SubscriptionBlockingPendingQueue inputPendingQueue;
 
   private final AtomicLong commitIdGenerator;
+  // record initial commit for outdated event detection
+  private final long initialCommitId;
 
   /** A queue containing a series of prefetched pollable {@link SubscriptionEvent}. */
   protected final PriorityBlockingQueue<SubscriptionEvent> prefetchingQueue;
@@ -113,6 +115,7 @@ public SubscriptionPrefetchingQueue(
     this.topicName = topicName;
     this.inputPendingQueue = inputPendingQueue;
     this.commitIdGenerator = commitIdGenerator;
+    this.initialCommitId = commitIdGenerator.get();
 
     this.prefetchingQueue = new PriorityBlockingQueue<>();
     this.inFlightEvents = new ConcurrentHashMap<>();
@@ -582,6 +585,24 @@ protected SubscriptionEvent generateSubscriptionPollErrorResponse(final String e
             INVALID_COMMIT_ID));
   }
 
+  protected SubscriptionEvent generateSubscriptionPollOutdatedErrorResponse() {
+    // consider non-critical by default, meaning the client can retry
+    return new SubscriptionEvent(
+        SubscriptionPollResponseType.ERROR.getType(),
+        ErrorPayload.OUTDATED_ERROR_PAYLOAD,
+        new SubscriptionCommitContext(
+            IoTDBDescriptor.getInstance().getConfig().getDataNodeId(),
+            PipeDataNodeAgent.runtime().getRebootTimes(),
+            topicName,
+            brokerId,
+            INVALID_COMMIT_ID));
+  }
+
+  public boolean isCommitContextOutdated(final SubscriptionCommitContext commitContext) {
+    return PipeDataNodeAgent.runtime().getRebootTimes() > commitContext.getRebootTimes()
+        || initialCommitId > commitContext.getCommitId();
+  }
+
   //////////////////////////// APIs provided for metric framework ////////////////////////////
 
   public String getPrefetchingQueueId() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTabletQueue.java

@@ -78,6 +78,16 @@ public SubscriptionEvent pollTablets(
         (key, ev) -> {
           // 1. Extract current event and check it
           if (Objects.isNull(ev)) {
+            if (isCommitContextOutdated(commitContext)) {
+              LOGGER.warn(
+                  "SubscriptionPrefetchingTabletQueue {} detected outdated poll request, consumer {}, commit context {}, offset {}",
+                  this,
+                  consumerId,
+                  commitContext,
+                  offset);
+              eventRef.set(generateSubscriptionPollOutdatedErrorResponse());
+              return null;
+            }
             final String errorMessage =
                 String.format(
                     "SubscriptionPrefetchingTabletQueue %s is currently not transferring any tablet to consumer %s, commit context: %s, offset: %s",

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingTsFileQueue.java

@@ -84,6 +84,16 @@ public SubscriptionEvent pollTsFile(
         (key, ev) -> {
           // 1. Extract current event and check it
           if (Objects.isNull(ev)) {
+            if (isCommitContextOutdated(commitContext)) {
+              LOGGER.warn(
+                  "SubscriptionPrefetchingTsFileQueue {} detected outdated poll request, consumer {}, commit context {}, writing offset {}",
+                  this,
+                  consumerId,
+                  commitContext,
+                  writingOffset);
+              eventRef.set(generateSubscriptionPollOutdatedErrorResponse());
+              return null;
+            }
             final String errorMessage =
                 String.format(
                     "SubscriptionPrefetchingTsFileQueue %s is currently not transferring any file to consumer %s, commit context: %s, writing offset: %s",

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java

@@ -425,14 +425,20 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo
                   final SubscriptionCommitContext commitContext = event.getCommitContext();
                   final SubscriptionPollResponse response = event.getCurrentResponse();
                   if (Objects.isNull(response)) {
+                    final boolean isOutdated =
+                        SubscriptionAgent.broker()
+                            .isCommitContextOutdated(event.getCommitContext());
                     LOGGER.warn(
-                        "Subscription: consumer {} poll null response for event {} with request: {}",
+                        "Subscription: consumer {} poll null response for event {} (outdated: {}) with request: {}",
                         consumerConfig,
                         event,
+                        isOutdated,
                         req.getRequest());
                     // nack
-                    SubscriptionAgent.broker()
-                        .commit(consumerConfig, Collections.singletonList(commitContext), true);
+                    if (!isOutdated) {
+                      SubscriptionAgent.broker()
+                          .commit(consumerConfig, Collections.singletonList(commitContext), true);
+                    }
                     return null;
                   }
 
@@ -462,24 +468,33 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo
                         req.getRequest());
                     return byteBuffer;
                   } catch (final Exception e) {
+                    final boolean isOutdated =
+                        SubscriptionAgent.broker()
+                            .isCommitContextOutdated(event.getCommitContext());
                     if (e instanceof SubscriptionPayloadExceedException) {
                       LOGGER.error(
-                          "Subscription: consumer {} poll excessive payload {} with request: {}, something unexpected happened with parameter configuration or payload control...",
+                          "Subscription: consumer {} poll excessive payload {} for event {} (outdated: {}) with request: {}, something unexpected happened with parameter configuration or payload control...",
                           consumerConfig,
                           response,
+                          event,
+                          isOutdated,
                           req.getRequest(),
                           e);
                     } else {
                       LOGGER.warn(
-                          "Subscription: consumer {} poll {} failed with request: {}",
+                          "Subscription: consumer {} poll {} for event {} (outdated: {}) failed with request: {}",
                           consumerConfig,
                           response,
+                          event,
+                          isOutdated,
                           req.getRequest(),
                           e);
                     }
                     // nack
-                    SubscriptionAgent.broker()
-                        .commit(consumerConfig, Collections.singletonList(commitContext), true);
+                    if (!isOutdated) {
+                      SubscriptionAgent.broker()
+                          .commit(consumerConfig, Collections.singletonList(commitContext), true);
+                    }
                     return null;
                   }
                 })