[iotdb] 01/01: Pipe: buffered events in processor stage can not be consumed by connector

[ro...@apache.org]

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

rong pushed a commit to branch IOTDB-6127
in repository https://gitbox.apache.org/repos/asf/iotdb.git

commit eff556382ae2c78c02cb7ef4d391b55563499216
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Sat Aug 26 23:45:48 2023 +0800

    Pipe: buffered events in processor stage can not be consumed by connector
---
 .../db/pipe/processor/PipeDoNothingProcessor.java  | 67 ----------------------
 .../pipe/task/connection/PipeEventCollector.java   | 22 +++++--
 .../db/pipe/task/stage/PipeTaskProcessorStage.java |  4 +-
 .../subtask/processor/PipeProcessorSubtask.java    | 11 ++--
 .../TsFileResourceProgressIndexTest.java           | 21 +++++++
 5 files changed, 47 insertions(+), 78 deletions(-)

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/PipeDoNothingProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/PipeDoNothingProcessor.java
deleted file mode 100644
index e1de3e13331..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/PipeDoNothingProcessor.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.pipe.processor;
-
-import org.apache.iotdb.pipe.api.PipeProcessor;
-import org.apache.iotdb.pipe.api.collector.EventCollector;
-import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
-import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
-import org.apache.iotdb.pipe.api.event.Event;
-import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
-import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
-
-import java.io.IOException;
-
-public class PipeDoNothingProcessor implements PipeProcessor {
-
-  @Override
-  public void validate(PipeParameterValidator validator) {
-    // do nothing
-  }
-
-  @Override
-  public void customize(
-      PipeParameters parameters, PipeProcessorRuntimeConfiguration configuration) {
-    // do nothing
-  }
-
-  @Override
-  public void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector)
-      throws IOException {
-    eventCollector.collect(tabletInsertionEvent);
-  }
-
-  @Override
-  public void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector)
-      throws IOException {
-    eventCollector.collect(tsFileInsertionEvent);
-  }
-
-  @Override
-  public void process(Event event, EventCollector eventCollector) throws IOException {
-    eventCollector.collect(event);
-  }
-
-  @Override
-  public void close() {
-    // do nothing
-  }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
index 8ec84529d08..bf57908c71b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java
@@ -23,16 +23,11 @@ import org.apache.iotdb.db.pipe.event.EnrichedEvent;
 import org.apache.iotdb.pipe.api.collector.EventCollector;
 import org.apache.iotdb.pipe.api.event.Event;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.LinkedList;
 import java.util.Queue;
 
 public class PipeEventCollector implements EventCollector {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(PipeEventCollector.class);
-
   private final BoundedBlockingPendingQueue<Event> pendingQueue;
 
   private final Queue<Event> bufferQueue;
@@ -64,4 +59,21 @@ public class PipeEventCollector implements EventCollector {
       bufferQueue.offer(event);
     }
   }
+
+  /**
+   * Try to collect buffered events into pending queue.
+   *
+   * @return true if there are still buffered events after this operation, false otherwise.
+   */
+  public synchronized boolean tryCollectBufferedEvents() {
+    while (!bufferQueue.isEmpty()) {
+      final Event bufferedEvent = bufferQueue.peek();
+      if (pendingQueue.waitedOffer(bufferedEvent)) {
+        bufferQueue.poll();
+      } else {
+        return true;
+      }
+    }
+    return false;
+  }
 }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
index d187bab7e59..894e7d14990 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/stage/PipeTaskProcessorStage.java
@@ -21,13 +21,13 @@ package org.apache.iotdb.db.pipe.task.stage;
 
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
 import org.apache.iotdb.commons.pipe.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.plugin.builtin.processor.DoNothingProcessor;
 import org.apache.iotdb.db.pipe.agent.PipeAgent;
 import org.apache.iotdb.db.pipe.config.constant.PipeProcessorConstant;
 import org.apache.iotdb.db.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
 import org.apache.iotdb.db.pipe.config.plugin.env.PipeTaskRuntimeEnvironment;
 import org.apache.iotdb.db.pipe.execution.executor.PipeProcessorSubtaskExecutor;
 import org.apache.iotdb.db.pipe.execution.executor.PipeSubtaskExecutorManager;
-import org.apache.iotdb.db.pipe.processor.PipeDoNothingProcessor;
 import org.apache.iotdb.db.pipe.task.connection.BoundedBlockingPendingQueue;
 import org.apache.iotdb.db.pipe.task.connection.EventSupplier;
 import org.apache.iotdb.db.pipe.task.connection.PipeEventCollector;
@@ -68,7 +68,7 @@ public class PipeTaskProcessorStage extends PipeTaskStage {
                     PipeProcessorConstant.PROCESSOR_KEY,
                     BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName())
                 .equals(BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName())
-            ? new PipeDoNothingProcessor()
+            ? new DoNothingProcessor()
             : PipeAgent.plugin().reflectProcessor(pipeProcessorParameters);
 
     // validate and customize should be called before createSubtask. this allows extractor exposing
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
index 0c7b96238c1..1e0feb2abeb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
@@ -22,9 +22,9 @@ package org.apache.iotdb.db.pipe.task.subtask.processor;
 import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
 import org.apache.iotdb.db.pipe.execution.scheduler.PipeSubtaskScheduler;
 import org.apache.iotdb.db.pipe.task.connection.EventSupplier;
+import org.apache.iotdb.db.pipe.task.connection.PipeEventCollector;
 import org.apache.iotdb.db.pipe.task.subtask.PipeSubtask;
 import org.apache.iotdb.pipe.api.PipeProcessor;
-import org.apache.iotdb.pipe.api.collector.EventCollector;
 import org.apache.iotdb.pipe.api.event.Event;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
@@ -47,7 +47,7 @@ public class PipeProcessorSubtask extends PipeSubtask {
 
   private final EventSupplier inputEventSupplier;
   private final PipeProcessor pipeProcessor;
-  private final EventCollector outputEventCollector;
+  private final PipeEventCollector outputEventCollector;
 
   private final AtomicBoolean isClosed;
 
@@ -55,7 +55,7 @@ public class PipeProcessorSubtask extends PipeSubtask {
       String taskID,
       EventSupplier inputEventSupplier,
       PipeProcessor pipeProcessor,
-      EventCollector outputEventCollector) {
+      PipeEventCollector outputEventCollector) {
     super(taskID);
     this.inputEventSupplier = inputEventSupplier;
     this.pipeProcessor = pipeProcessor;
@@ -89,7 +89,10 @@ public class PipeProcessorSubtask extends PipeSubtask {
     // Record the last event for retry when exception occurs
     lastEvent = event;
     if (event == null) {
-      return false;
+      // Though there is no event to process, there may still be some buffered events
+      // in the outputEventCollector. Return true if there are still buffered events,
+      // false otherwise.
+      return outputEventCollector.tryCollectBufferedEvents();
     }
 
     try {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java
index 62b9caa2f35..0dd80260cee 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.storageengine.dataregion;
 import org.apache.iotdb.commons.consensus.index.ProgressIndex;
 import org.apache.iotdb.commons.consensus.index.ProgressIndexType;
 import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -197,4 +198,24 @@ public class TsFileResourceProgressIndexTest {
       throw new UnsupportedOperationException("method not implemented.");
     }
   }
+
+  @Test
+  public void testHybridProgressIndex() {
+    final IoTProgressIndex ioTProgressIndex = new IoTProgressIndex(1, 123L);
+    final RecoverProgressIndex recoverProgressIndex =
+        new RecoverProgressIndex(1, new SimpleProgressIndex(2, 2));
+    final HybridProgressIndex hybridProgressIndex = new HybridProgressIndex();
+
+    hybridProgressIndex.updateToMinimumIsAfterProgressIndex(ioTProgressIndex);
+    hybridProgressIndex.updateToMinimumIsAfterProgressIndex(recoverProgressIndex);
+
+    Assert.assertTrue(hybridProgressIndex.isAfter(new IoTProgressIndex(1, 100L)));
+    Assert.assertTrue(
+        hybridProgressIndex.isAfter(new RecoverProgressIndex(1, new SimpleProgressIndex(1, 2))));
+
+    Assert.assertFalse(hybridProgressIndex.isAfter(new IoTProgressIndex(1, 200L)));
+    Assert.assertFalse(hybridProgressIndex.isAfter(new IoTProgressIndex(2, 200L)));
+    Assert.assertFalse(
+        hybridProgressIndex.isAfter(new RecoverProgressIndex(1, new SimpleProgressIndex(2, 21))));
+  }
 }