(iotdb) branch dev/1.3 updated: Subscription: detect outdated subscription event for tsfile message (#15430) (#15460)

[rong]

This is an automated email from the ASF dual-hosted git repository.

rong pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/dev/1.3 by this push:
     new 37fced1ce05 Subscription: detect outdated subscription event for 
tsfile message (#15430) (#15460)
37fced1ce05 is described below

commit 37fced1ce05b9cf86d23964a41e76557b948e629
Author: VGalaxies <vgalax...@apache.org>
AuthorDate: Wed May 7 15:15:36 2025 +0800

    Subscription: detect outdated subscription event for tsfile message 
(#15430) (#15460)
---
 .../subscription/consumer/SubscriptionConsumer.java       | 15 ++++++---------
 1 file changed, 6 insertions(+), 9 deletions(-)

diff --git 
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
 
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
index 324797967fb..8cbe1b23cb3 100644
--- 
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
+++ 
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/SubscriptionConsumer.java
@@ -711,7 +711,7 @@ abstract class SubscriptionConsumer implements 
AutoCloseable {
     final Path filePath = getFilePath(commitContext, topicName, fileName, 
true, true);
     final File file = filePath.toFile();
     try (final RandomAccessFile fileWriter = new RandomAccessFile(file, "rw")) 
{
-      return Optional.of(pollFileInternal(commitContext, fileName, file, 
fileWriter, timer));
+      return pollFileInternal(commitContext, fileName, file, fileWriter, 
timer);
     } catch (final Exception e) {
       if (!(e instanceof SubscriptionPollTimeoutException)) {
         inFlightFilesCommitContextSet.remove(commitContext);
@@ -724,7 +724,7 @@ abstract class SubscriptionConsumer implements 
AutoCloseable {
     }
   }
 
-  private SubscriptionMessage pollFileInternal(
+  private Optional<SubscriptionMessage> pollFileInternal(
       final SubscriptionCommitContext commitContext,
       final String rawFileName,
       final File file,
@@ -757,13 +757,10 @@ abstract class SubscriptionConsumer implements 
AutoCloseable {
       final List<SubscriptionPollResponse> responses =
           pollFileInternal(commitContext, writingOffset, timer.remainingMs());
 
-      // It's agreed that the server will always return at least one response, 
even in case of
-      // failure.
+      // If responses is empty, it means that some outdated subscription 
events may be being polled,
+      // so just return.
       if (responses.isEmpty()) {
-        final String errorMessage =
-            String.format("SubscriptionConsumer %s poll empty response", this);
-        LOGGER.warn(errorMessage);
-        throw new SubscriptionRuntimeNonCriticalException(errorMessage);
+        return Optional.empty();
       }
 
       // only one SubscriptionEvent polled currently
@@ -872,7 +869,7 @@ abstract class SubscriptionConsumer implements 
AutoCloseable {
 
             // generate subscription message
             inFlightFilesCommitContextSet.remove(commitContext);
-            return new SubscriptionMessage(commitContext, 
file.getAbsolutePath());
+            return Optional.of(new SubscriptionMessage(commitContext, 
file.getAbsolutePath()));
           }
         case ERROR:
           {