Caideyipi commented on code in PR #12355: URL: https://github.com/apache/iotdb/pull/12355#discussion_r1604662755
########## iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.consensus.pipe; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.consensus.index.ComparableConsensusRequest; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; +import org.apache.iotdb.commons.pipe.task.meta.PipeStatus; +import org.apache.iotdb.consensus.IStateMachine; +import org.apache.iotdb.consensus.common.DataSet; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.common.request.IConsensusRequest; +import org.apache.iotdb.consensus.config.PipeConsensusConfig; +import org.apache.iotdb.consensus.exception.ConsensusGroupModifyPeerException; +import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeManager; +import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName; +import org.apache.iotdb.consensus.pipe.consensuspipe.ProgressIndexManager; + +import com.google.common.collect.ImmutableMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; + +import static org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeManager.getConsensusPipeName; + +/** PipeConsensusServerImpl is a consensus server implementation for pipe consensus. */ +public class PipeConsensusServerImpl { + private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusServerImpl.class); + + private final Peer thisNode; + private final IStateMachine stateMachine; + private final Lock stateMachineLock = new ReentrantLock(); + private final PipeConsensusPeerManager peerManager; + private final PipeConsensusConfig config; + private final AtomicBoolean active; + private final AtomicBoolean isStarted; + private final String consensusGroupId; + private final ConsensusPipeManager consensusPipeManager; + private final ProgressIndexManager progressIndexManager; + + private ProgressIndex cachedProgressIndex = MinimumProgressIndex.INSTANCE; + + public PipeConsensusServerImpl( + Peer thisNode, + IStateMachine stateMachine, + String storageDir, + List<Peer> configuration, + PipeConsensusConfig config, + ConsensusPipeManager consensusPipeManager) + throws IOException { + this.thisNode = thisNode; + this.stateMachine = stateMachine; + this.peerManager = new PipeConsensusPeerManager(storageDir, configuration); + this.config = config; + this.active = new AtomicBoolean(true); + this.isStarted = new AtomicBoolean(false); + this.consensusGroupId = thisNode.getGroupId().toString(); + this.consensusPipeManager = consensusPipeManager; + this.progressIndexManager = config.getPipe().getProgressIndexManager(); + + if (configuration.isEmpty()) { + peerManager.recover(); + } else { + // create consensus pipes + configuration.remove(thisNode); + final List<Peer> successfulPips = createConsensusPipes(configuration); Review Comment: successfulPipes ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java: ########## @@ -0,0 +1,530 @@ +/* + * 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.connector.protocol.pipeconsensus; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.connector.protocol.IoTDBConnector; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.consensus.pipe.client.AsyncPipeConsensusServiceClient; +import org.apache.iotdb.consensus.pipe.client.manager.PipeConsensusAsyncClientManager; +import org.apache.iotdb.consensus.pipe.thrift.TCommitId; +import org.apache.iotdb.consensus.pipe.thrift.TPipeConsensusTransferReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletBatchEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletInsertNodeEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletRawEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTsFileInsertionEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.builder.PipeConsensusAsyncBatchReqBuilder; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletBinaryReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletInsertNodeReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletRawReq; +import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration; +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 org.apache.iotdb.pipe.api.exception.PipeException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.Comparator; +import java.util.Iterator; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +// TODO: 改造 handler，onComplete 的优化 + onComplete 加上出队逻辑 +// TODO: 改造 batch 协议 +// TODO: 改造 tsFile 传送协议 +public class PipeConsensusAsyncConnector extends IoTDBConnector { + + private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusAsyncConnector.class); + + private static final String ENQUEUE_EXCEPTION_MSG = + "Timeout: PipeConsensusConnector offers an event into transferBuffer failed, because transferBuffer is full"; + + private static final String THRIFT_ERROR_FORMATTER_WITHOUT_ENDPOINT = + "Failed to borrow client from client pool or exception occurred " + + "when sending to receiver."; + + private static final String THRIFT_ERROR_FORMATTER_WITH_ENDPOINT = + "Failed to borrow client from client pool or exception occurred " + + "when sending to receiver %s:%s."; + + private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig(); + + private final PriorityBlockingQueue<Event> retryEventQueue = + new PriorityBlockingQueue<>( + 11, + Comparator.comparing( + e -> + // Non-enriched events will be put at the front of the queue, + // because they are more likely to be lost and need to be retried first. + e instanceof EnrichedEvent ? ((EnrichedEvent) e).getCommitId() : 0)); + + private final BlockingQueue<Event> transferBuffer = + new LinkedBlockingDeque<>(COMMON_CONFIG.getPipeConsensusEventBufferSize()); + + private final AtomicBoolean isClosed = new AtomicBoolean(false); + + private final AtomicInteger alreadySentEventsInTransferBuffer = new AtomicInteger(0); + + private PipeConsensusSyncConnector retryConnector; + + private PipeConsensusAsyncClientManager asyncTransferClientManager; + + private PipeConsensusAsyncBatchReqBuilder tabletBatchBuilder; + + @Override + public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) + throws Exception { + super.customize(parameters, configuration); + + // In PipeConsensus, one pipeConsensusTask corresponds to a pipeConsensusConnector. Thus, + // `nodeUrls` here actually is a singletonList that contains one peer's TEndPoint. But here we + // retain the implementation of list to cope with possible future expansion + retryConnector = new PipeConsensusSyncConnector(nodeUrls); + retryConnector.customize(parameters, configuration); + asyncTransferClientManager = PipeConsensusAsyncClientManager.getInstance(); + + if (isTabletBatchModeEnabled) { + tabletBatchBuilder = new PipeConsensusAsyncBatchReqBuilder(parameters); + } + + // currently, tablet batch is false by default in PipeConsensus; + isTabletBatchModeEnabled = false; + } + + /** Add an event to transferBuffer, whose events will be asynchronizedly transfer to receiver. */ + private boolean addEvent2Buffer(Event event) { + try { + LOGGER.info( + "Debug only: no.{} event added to connector buffer", + ((EnrichedEvent) event).getCommitId()); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "PipeConsensus connector: one event enqueue, queue size = {}, limit size = {}", + transferBuffer.size(), + COMMON_CONFIG.getPipeConsensusEventBufferSize()); + } + boolean result = + transferBuffer.offer( + event, COMMON_CONFIG.getPipeConsensusEventEnqueueTimeoutInMs(), TimeUnit.SECONDS); + // add reference + if (result) { + ((EnrichedEvent) event).increaseReferenceCount(PipeConsensusAsyncConnector.class.getName()); + } + return result; + } catch (InterruptedException e) { + LOGGER.info("PipeConsensusConnector transferBuffer queue offer is interrupted.", e); + Thread.currentThread().interrupt(); + return false; + } + } + + /** + * if one event is successfully processed by receiver in PipeConsensus, we will remove this event + * from transferBuffer in order to transfer other event. + */ + public synchronized void removeEventFromBuffer(Event event) { Review Comment: Why are there double "synchronized".... ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java: ########## @@ -0,0 +1,530 @@ +/* + * 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.connector.protocol.pipeconsensus; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.connector.protocol.IoTDBConnector; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.consensus.pipe.client.AsyncPipeConsensusServiceClient; +import org.apache.iotdb.consensus.pipe.client.manager.PipeConsensusAsyncClientManager; +import org.apache.iotdb.consensus.pipe.thrift.TCommitId; +import org.apache.iotdb.consensus.pipe.thrift.TPipeConsensusTransferReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletBatchEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletInsertNodeEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletRawEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTsFileInsertionEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.builder.PipeConsensusAsyncBatchReqBuilder; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletBinaryReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletInsertNodeReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletRawReq; +import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration; +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 org.apache.iotdb.pipe.api.exception.PipeException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.Comparator; +import java.util.Iterator; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +// TODO: 改造 handler，onComplete 的优化 + onComplete 加上出队逻辑 +// TODO: 改造 batch 协议 +// TODO: 改造 tsFile 传送协议 +public class PipeConsensusAsyncConnector extends IoTDBConnector { + + private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusAsyncConnector.class); + + private static final String ENQUEUE_EXCEPTION_MSG = + "Timeout: PipeConsensusConnector offers an event into transferBuffer failed, because transferBuffer is full"; + + private static final String THRIFT_ERROR_FORMATTER_WITHOUT_ENDPOINT = + "Failed to borrow client from client pool or exception occurred " + + "when sending to receiver."; + + private static final String THRIFT_ERROR_FORMATTER_WITH_ENDPOINT = + "Failed to borrow client from client pool or exception occurred " + + "when sending to receiver %s:%s."; + + private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig(); + + private final PriorityBlockingQueue<Event> retryEventQueue = + new PriorityBlockingQueue<>( + 11, + Comparator.comparing( + e -> + // Non-enriched events will be put at the front of the queue, + // because they are more likely to be lost and need to be retried first. + e instanceof EnrichedEvent ? ((EnrichedEvent) e).getCommitId() : 0)); + + private final BlockingQueue<Event> transferBuffer = + new LinkedBlockingDeque<>(COMMON_CONFIG.getPipeConsensusEventBufferSize()); + + private final AtomicBoolean isClosed = new AtomicBoolean(false); + + private final AtomicInteger alreadySentEventsInTransferBuffer = new AtomicInteger(0); + + private PipeConsensusSyncConnector retryConnector; + + private PipeConsensusAsyncClientManager asyncTransferClientManager; + + private PipeConsensusAsyncBatchReqBuilder tabletBatchBuilder; + + @Override + public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) + throws Exception { + super.customize(parameters, configuration); + + // In PipeConsensus, one pipeConsensusTask corresponds to a pipeConsensusConnector. Thus, + // `nodeUrls` here actually is a singletonList that contains one peer's TEndPoint. But here we + // retain the implementation of list to cope with possible future expansion + retryConnector = new PipeConsensusSyncConnector(nodeUrls); + retryConnector.customize(parameters, configuration); + asyncTransferClientManager = PipeConsensusAsyncClientManager.getInstance(); + + if (isTabletBatchModeEnabled) { + tabletBatchBuilder = new PipeConsensusAsyncBatchReqBuilder(parameters); + } + + // currently, tablet batch is false by default in PipeConsensus; + isTabletBatchModeEnabled = false; + } + + /** Add an event to transferBuffer, whose events will be asynchronizedly transfer to receiver. */ + private boolean addEvent2Buffer(Event event) { + try { + LOGGER.info( + "Debug only: no.{} event added to connector buffer", + ((EnrichedEvent) event).getCommitId()); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "PipeConsensus connector: one event enqueue, queue size = {}, limit size = {}", + transferBuffer.size(), + COMMON_CONFIG.getPipeConsensusEventBufferSize()); + } + boolean result = + transferBuffer.offer( + event, COMMON_CONFIG.getPipeConsensusEventEnqueueTimeoutInMs(), TimeUnit.SECONDS); Review Comment: Why are there "Ms" and “SECONDS”？ ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java: ########## @@ -0,0 +1,530 @@ +/* + * 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.connector.protocol.pipeconsensus; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.pipe.connector.protocol.IoTDBConnector; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.consensus.pipe.client.AsyncPipeConsensusServiceClient; +import org.apache.iotdb.consensus.pipe.client.manager.PipeConsensusAsyncClientManager; +import org.apache.iotdb.consensus.pipe.thrift.TCommitId; +import org.apache.iotdb.consensus.pipe.thrift.TPipeConsensusTransferReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletBatchEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletInsertNodeEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTabletRawEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.handler.PipeConsensusTsFileInsertionEventHandler; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.builder.PipeConsensusAsyncBatchReqBuilder; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletBinaryReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletInsertNodeReq; +import org.apache.iotdb.db.pipe.connector.protocol.pipeconsensus.payload.request.PipeConsensusTabletRawReq; +import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration; +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 org.apache.iotdb.pipe.api.exception.PipeException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.Comparator; +import java.util.Iterator; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +// TODO: 改造 handler，onComplete 的优化 + onComplete 加上出队逻辑 +// TODO: 改造 batch 协议 +// TODO: 改造 tsFile 传送协议 +public class PipeConsensusAsyncConnector extends IoTDBConnector { + + private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusAsyncConnector.class); + + private static final String ENQUEUE_EXCEPTION_MSG = + "Timeout: PipeConsensusConnector offers an event into transferBuffer failed, because transferBuffer is full"; + + private static final String THRIFT_ERROR_FORMATTER_WITHOUT_ENDPOINT = + "Failed to borrow client from client pool or exception occurred " + + "when sending to receiver."; + + private static final String THRIFT_ERROR_FORMATTER_WITH_ENDPOINT = + "Failed to borrow client from client pool or exception occurred " + + "when sending to receiver %s:%s."; + + private static final CommonConfig COMMON_CONFIG = CommonDescriptor.getInstance().getConfig(); + + private final PriorityBlockingQueue<Event> retryEventQueue = + new PriorityBlockingQueue<>( + 11, + Comparator.comparing( + e -> + // Non-enriched events will be put at the front of the queue, + // because they are more likely to be lost and need to be retried first. + e instanceof EnrichedEvent ? ((EnrichedEvent) e).getCommitId() : 0)); + + private final BlockingQueue<Event> transferBuffer = + new LinkedBlockingDeque<>(COMMON_CONFIG.getPipeConsensusEventBufferSize()); + + private final AtomicBoolean isClosed = new AtomicBoolean(false); + + private final AtomicInteger alreadySentEventsInTransferBuffer = new AtomicInteger(0); + + private PipeConsensusSyncConnector retryConnector; + + private PipeConsensusAsyncClientManager asyncTransferClientManager; + + private PipeConsensusAsyncBatchReqBuilder tabletBatchBuilder; + + @Override + public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) + throws Exception { + super.customize(parameters, configuration); + + // In PipeConsensus, one pipeConsensusTask corresponds to a pipeConsensusConnector. Thus, + // `nodeUrls` here actually is a singletonList that contains one peer's TEndPoint. But here we + // retain the implementation of list to cope with possible future expansion + retryConnector = new PipeConsensusSyncConnector(nodeUrls); + retryConnector.customize(parameters, configuration); + asyncTransferClientManager = PipeConsensusAsyncClientManager.getInstance(); + + if (isTabletBatchModeEnabled) { + tabletBatchBuilder = new PipeConsensusAsyncBatchReqBuilder(parameters); + } + + // currently, tablet batch is false by default in PipeConsensus; + isTabletBatchModeEnabled = false; + } + + /** Add an event to transferBuffer, whose events will be asynchronizedly transfer to receiver. */ + private boolean addEvent2Buffer(Event event) { + try { + LOGGER.info( + "Debug only: no.{} event added to connector buffer", + ((EnrichedEvent) event).getCommitId()); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "PipeConsensus connector: one event enqueue, queue size = {}, limit size = {}", + transferBuffer.size(), + COMMON_CONFIG.getPipeConsensusEventBufferSize()); + } + boolean result = + transferBuffer.offer( + event, COMMON_CONFIG.getPipeConsensusEventEnqueueTimeoutInMs(), TimeUnit.SECONDS); + // add reference + if (result) { + ((EnrichedEvent) event).increaseReferenceCount(PipeConsensusAsyncConnector.class.getName()); + } + return result; + } catch (InterruptedException e) { + LOGGER.info("PipeConsensusConnector transferBuffer queue offer is interrupted.", e); + Thread.currentThread().interrupt(); + return false; + } + } + + /** + * if one event is successfully processed by receiver in PipeConsensus, we will remove this event + * from transferBuffer in order to transfer other event. + */ + public synchronized void removeEventFromBuffer(Event event) { + synchronized (this) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug( + "PipeConsensus connector: one event removed from queue, queue size = {}, limit size = {}", + transferBuffer.size(), + COMMON_CONFIG.getPipeConsensusEventBufferSize()); + } + Iterator<Event> iterator = transferBuffer.iterator(); + Event current = iterator.next(); + while (!current.equals(event) && iterator.hasNext()) { + current = iterator.next(); + } + iterator.remove(); + // decrease reference count + ((EnrichedEvent) event) + .decreaseReferenceCount(PipeConsensusAsyncConnector.class.getName(), true); + // decrease alreadySentEventsCounts + alreadySentEventsInTransferBuffer.decrementAndGet(); + } + } + + @Override + public void handshake() throws Exception { + // do nothing + // PipeConsensus doesn't need to do handshake, since nodes in same consensusGroup/cluster + // usually have same configuration. + } + + @Override + public void heartbeat() throws Exception { + // do nothing + } + + @Override + public void transfer(TabletInsertionEvent tabletInsertionEvent) throws Exception { + boolean enqueueResult = addEvent2Buffer(tabletInsertionEvent); + if (!enqueueResult) { + throw new PipeException(ENQUEUE_EXCEPTION_MSG); + } + + syncTransferQueuedEventsIfNecessary(); + + if (!(tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) + && !(tabletInsertionEvent instanceof PipeRawTabletInsertionEvent)) { + LOGGER.warn( + "IoTDBThriftAsyncConnector only support PipeInsertNodeTabletInsertionEvent and PipeRawTabletInsertionEvent. " + + "Current event: {}.", + tabletInsertionEvent); + return; + } + + // batch transfer tablets. + if (isTabletBatchModeEnabled) { + if (tabletBatchBuilder.onEvent(tabletInsertionEvent)) { + final PipeConsensusTabletBatchEventHandler pipeConsensusTabletBatchEventHandler = + new PipeConsensusTabletBatchEventHandler(tabletBatchBuilder, this); + + transfer(pipeConsensusTabletBatchEventHandler); + + tabletBatchBuilder.onSuccess(); + } + } else { + TCommitId tCommitId; + if (tabletInsertionEvent instanceof PipeInsertNodeTabletInsertionEvent) { + final PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletInsertionEvent = + (PipeInsertNodeTabletInsertionEvent) tabletInsertionEvent; + tCommitId = + new TCommitId( + pipeInsertNodeTabletInsertionEvent.getCommitId(), + pipeInsertNodeTabletInsertionEvent.getRebootTimes()); + + // We increase the reference count for this event to determine if the event may be released. + if (!pipeInsertNodeTabletInsertionEvent.increaseReferenceCount( + PipeConsensusAsyncConnector.class.getName())) { + pipeInsertNodeTabletInsertionEvent.decreaseReferenceCount( + PipeConsensusAsyncConnector.class.getName(), false); + return; + } + + final InsertNode insertNode = + pipeInsertNodeTabletInsertionEvent.getInsertNodeViaCacheIfPossible(); + final TPipeConsensusTransferReq pipeConsensusTransferReq = + Objects.isNull(insertNode) + ? PipeConsensusTabletBinaryReq.toTPipeConsensusTransferReq( + pipeInsertNodeTabletInsertionEvent.getByteBuffer(), tCommitId) + : PipeConsensusTabletInsertNodeReq.toTPipeConsensusTransferReq( + insertNode, tCommitId); + final PipeConsensusTabletInsertNodeEventHandler pipeConsensusInsertNodeReqHandler = + new PipeConsensusTabletInsertNodeEventHandler( + pipeInsertNodeTabletInsertionEvent, pipeConsensusTransferReq, this); + + transfer(pipeConsensusInsertNodeReqHandler); + } else { // tabletInsertionEvent instanceof PipeRawTabletInsertionEvent + final PipeRawTabletInsertionEvent pipeRawTabletInsertionEvent = + (PipeRawTabletInsertionEvent) tabletInsertionEvent; + tCommitId = + new TCommitId( + pipeRawTabletInsertionEvent.getCommitId(), + pipeRawTabletInsertionEvent.getRebootTimes()); + + // We increase the reference count for this event to determine if the event may be released. + if (!pipeRawTabletInsertionEvent.increaseReferenceCount( + PipeConsensusAsyncConnector.class.getName())) { + pipeRawTabletInsertionEvent.decreaseReferenceCount( + PipeConsensusAsyncConnector.class.getName(), false); + return; + } + + final PipeConsensusTabletRawReq pipeConsensusTabletRawReq = + PipeConsensusTabletRawReq.toTPipeConsensusTransferReq( + pipeRawTabletInsertionEvent.convertToTablet(), + pipeRawTabletInsertionEvent.isAligned(), + tCommitId); + final PipeConsensusTabletRawEventHandler pipeConsensusTabletRawEventHandler = + new PipeConsensusTabletRawEventHandler( + pipeRawTabletInsertionEvent, pipeConsensusTabletRawReq, this); + + transfer(pipeConsensusTabletRawEventHandler); + } + } + } + + private void transfer( + final PipeConsensusTabletBatchEventHandler pipeConsensusTabletBatchEventHandler) { + AsyncPipeConsensusServiceClient client = null; + try { + client = asyncTransferClientManager.borrowClient(getFollowerUrl()); + pipeConsensusTabletBatchEventHandler.transfer(client); + } catch (final Exception ex) { + logOnClientException(client, ex); + pipeConsensusTabletBatchEventHandler.onError(ex); + } + } + + private void transfer( + final PipeConsensusTabletInsertNodeEventHandler pipeConsensusInsertNodeReqHandler) { + AsyncPipeConsensusServiceClient client = null; + try { + client = asyncTransferClientManager.borrowClient(getFollowerUrl()); + pipeConsensusInsertNodeReqHandler.transfer(client); + } catch (final Exception ex) { + logOnClientException(client, ex); + pipeConsensusInsertNodeReqHandler.onError(ex); + } + } + + private void transfer(final PipeConsensusTabletRawEventHandler pipeConsensusTabletReqHandler) { + AsyncPipeConsensusServiceClient client = null; + try { + client = asyncTransferClientManager.borrowClient(getFollowerUrl()); + pipeConsensusTabletReqHandler.transfer(client); + } catch (final Exception ex) { + logOnClientException(client, ex); + pipeConsensusTabletReqHandler.onError(ex); + } + } + + @Override + public void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception { + boolean enqueueResult = addEvent2Buffer(tsFileInsertionEvent); + if (!enqueueResult) { + throw new PipeException(ENQUEUE_EXCEPTION_MSG); + } + + syncTransferQueuedEventsIfNecessary(); + transferBatchedEventsIfNecessary(); + + if (!(tsFileInsertionEvent instanceof PipeTsFileInsertionEvent)) { + LOGGER.warn( + "PipeConsensusAsyncConnector only support PipeTsFileInsertionEvent. Current event: {}.", + tsFileInsertionEvent); + return; + } + + final PipeTsFileInsertionEvent pipeTsFileInsertionEvent = + (PipeTsFileInsertionEvent) tsFileInsertionEvent; + TCommitId tCommitId = + new TCommitId( + pipeTsFileInsertionEvent.getCommitId(), pipeTsFileInsertionEvent.getRebootTimes()); + // We increase the reference count for this event to determine if the event may be released. + if (!pipeTsFileInsertionEvent.increaseReferenceCount( + PipeConsensusAsyncConnector.class.getName())) { + pipeTsFileInsertionEvent.decreaseReferenceCount( + PipeConsensusAsyncConnector.class.getName(), false); + return; + } + + // Just in case. To avoid the case that exception occurred when constructing the handler. + if (!pipeTsFileInsertionEvent.getTsFile().exists()) { + throw new FileNotFoundException(pipeTsFileInsertionEvent.getTsFile().getAbsolutePath()); + } + + final PipeConsensusTsFileInsertionEventHandler pipeConsensusTsFileInsertionEventHandler = + new PipeConsensusTsFileInsertionEventHandler(pipeTsFileInsertionEvent, this, tCommitId); + + transfer(pipeConsensusTsFileInsertionEventHandler); + } + + private void transfer( + final PipeConsensusTsFileInsertionEventHandler pipeConsensusTsFileInsertionEventHandler) { + AsyncPipeConsensusServiceClient client = null; + try { + client = asyncTransferClientManager.borrowClient(getFollowerUrl()); + pipeConsensusTsFileInsertionEventHandler.transfer(client); + } catch (final Exception ex) { + logOnClientException(client, ex); + pipeConsensusTsFileInsertionEventHandler.onError(ex); + } + } + + /** + * PipeConsensus only need transfer heartbeat event here. And heartbeat event doesn't need to be + * added to transferBuffer. + */ + @Override + public void transfer(Event event) throws Exception { + syncTransferQueuedEventsIfNecessary(); + transferBatchedEventsIfNecessary(); + + if (!(event instanceof PipeHeartbeatEvent)) { + LOGGER.warn( + "PipeConsensusAsyncConnector does not support transferring generic event: {}.", event); + return; + } + + retryConnector.transfer(event); + } + + /** Try its best to commit data in order. Flush can also be a trigger to transfer batched data. */ + private void transferBatchedEventsIfNecessary() throws IOException { + if (!isTabletBatchModeEnabled || tabletBatchBuilder.isEmpty()) { + return; + } + + transfer(new PipeConsensusTabletBatchEventHandler(tabletBatchBuilder, this)); + + tabletBatchBuilder.onSuccess(); + } + + /** + * Transfer queued {@link Event}s which are waiting for retry. + * + * @throws Exception if an error occurs. The error will be handled by pipe framework, which will + * retry the {@link Event} and mark the {@link Event} as failure and stop the pipe if the + * retry times exceeds the threshold. TODO: pipe 框架对于 Consensus 改成无限重试，而不是超过次数后 停止 Review Comment: Pipe will retry infinite times because the stopped pipe will be restarted in meta sync. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
