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 <[email protected]> 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)))); + } }
