Pengzna commented on code in PR #12355: URL: https://github.com/apache/iotdb/pull/12355#discussion_r1617095725
########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java: ########## @@ -0,0 +1,585 @@ +/* + * 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.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.client.IClientManager; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant; +import org.apache.iotdb.commons.pipe.connector.protocol.IoTDBConnector; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.consensus.pipe.PipeConsensus; +import org.apache.iotdb.consensus.pipe.client.AsyncPipeConsensusServiceClient; +import org.apache.iotdb.consensus.pipe.thrift.TCommitId; +import org.apache.iotdb.consensus.pipe.thrift.TPipeConsensusTransferReq; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; +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.Iterator; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +// TODO: Optimize the network and disk io for TsFile onComplete +// TODO: support Tablet Batch +public class PipeConsensusAsyncConnector extends IoTDBConnector { + + private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusAsyncConnector.class); + + private static final String CUSTOMIZE_EXCEPTION_MSG = + "Failed to customize pipeConsensusAsyncConnector because there isn't consensusGroupId passed by. Please check your construct parameters!"; + + 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 BlockingQueue<Event> retryEventQueue = new LinkedBlockingQueue<>(); + + 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 final int thisDataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); + + private String consensusGroupId; + + private PipeConsensusSyncConnector retryConnector; + + private IClientManager<TEndPoint, AsyncPipeConsensusServiceClient> asyncTransferClientManager; + + private PipeConsensusAsyncBatchReqBuilder tabletBatchBuilder; + + @Override + public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) + throws Exception { + super.customize(parameters, configuration); + + // Get consensusGroupId from parameters passed by PipeConsensusImpl + consensusGroupId = + parameters.getStringOrDefault(PipeConnectorConstant.CONNECTOR_CONSENSUS_GROUP_ID_KEY, null); + if (Objects.isNull(consensusGroupId)) { + throw new PipeException(CUSTOMIZE_EXCEPTION_MSG); + } + + // 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, consensusGroupId, thisDataNodeId); + retryConnector.customize(parameters, configuration); + asyncTransferClientManager = + ((PipeConsensus) DataRegionConsensusImpl.getInstance()).getAsyncClientManager(); + + if (isTabletBatchModeEnabled) { + tabletBatchBuilder = + new PipeConsensusAsyncBatchReqBuilder( + parameters, + new TConsensusGroupId( + TConsensusGroupType.DataRegion, Integer.parseInt(consensusGroupId)), + thisDataNodeId); + } + + // 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( + "PipeConsensus-ConsensusGroup-{}: no.{} event-{} added to connector buffer", + consensusGroupId, + ((EnrichedEvent) event).getCommitId(), + event); + 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.MILLISECONDS); + // 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) { + LOGGER.info( + "PipeConsensus-ConsensusGroup-{}: one event-{} successfully received by the follower, will be removed from queue, queue size = {}, limit size = {}", + consensusGroupId, + event, + 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) Review Comment: Good catch. fixed ########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java: ########## @@ -0,0 +1,585 @@ +/* + * 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.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.client.IClientManager; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.index.ProgressIndex; +import org.apache.iotdb.commons.pipe.config.constant.PipeConnectorConstant; +import org.apache.iotdb.commons.pipe.connector.protocol.IoTDBConnector; +import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.consensus.pipe.PipeConsensus; +import org.apache.iotdb.consensus.pipe.client.AsyncPipeConsensusServiceClient; +import org.apache.iotdb.consensus.pipe.thrift.TCommitId; +import org.apache.iotdb.consensus.pipe.thrift.TPipeConsensusTransferReq; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; +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.Iterator; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +// TODO: Optimize the network and disk io for TsFile onComplete +// TODO: support Tablet Batch +public class PipeConsensusAsyncConnector extends IoTDBConnector { + + private static final Logger LOGGER = LoggerFactory.getLogger(PipeConsensusAsyncConnector.class); + + private static final String CUSTOMIZE_EXCEPTION_MSG = + "Failed to customize pipeConsensusAsyncConnector because there isn't consensusGroupId passed by. Please check your construct parameters!"; + + 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 BlockingQueue<Event> retryEventQueue = new LinkedBlockingQueue<>(); + + 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 final int thisDataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); + + private String consensusGroupId; + + private PipeConsensusSyncConnector retryConnector; + + private IClientManager<TEndPoint, AsyncPipeConsensusServiceClient> asyncTransferClientManager; + + private PipeConsensusAsyncBatchReqBuilder tabletBatchBuilder; + + @Override + public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) + throws Exception { + super.customize(parameters, configuration); + + // Get consensusGroupId from parameters passed by PipeConsensusImpl + consensusGroupId = + parameters.getStringOrDefault(PipeConnectorConstant.CONNECTOR_CONSENSUS_GROUP_ID_KEY, null); + if (Objects.isNull(consensusGroupId)) { + throw new PipeException(CUSTOMIZE_EXCEPTION_MSG); + } + + // 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, consensusGroupId, thisDataNodeId); + retryConnector.customize(parameters, configuration); + asyncTransferClientManager = + ((PipeConsensus) DataRegionConsensusImpl.getInstance()).getAsyncClientManager(); + + if (isTabletBatchModeEnabled) { + tabletBatchBuilder = + new PipeConsensusAsyncBatchReqBuilder( + parameters, + new TConsensusGroupId( + TConsensusGroupType.DataRegion, Integer.parseInt(consensusGroupId)), + thisDataNodeId); + } + + // 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( + "PipeConsensus-ConsensusGroup-{}: no.{} event-{} added to connector buffer", + consensusGroupId, + ((EnrichedEvent) event).getCommitId(), + event); + 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.MILLISECONDS); + // 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) { + LOGGER.info( + "PipeConsensus-ConsensusGroup-{}: one event-{} successfully received by the follower, will be removed from queue, queue size = {}, limit size = {}", + consensusGroupId, + event, + 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; + TConsensusGroupId tConsensusGroupId = + new TConsensusGroupId(TConsensusGroupType.DataRegion, Integer.parseInt(consensusGroupId)); + 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 ProgressIndex progressIndex = pipeInsertNodeTabletInsertionEvent.getProgressIndex(); + final TPipeConsensusTransferReq pipeConsensusTransferReq = + Objects.isNull(insertNode) + ? PipeConsensusTabletBinaryReq.toTPipeConsensusTransferReq( + pipeInsertNodeTabletInsertionEvent.getByteBuffer(), + tCommitId, + tConsensusGroupId, + progressIndex, + thisDataNodeId) + : PipeConsensusTabletInsertNodeReq.toTPipeConsensusTransferReq( + insertNode, tCommitId, tConsensusGroupId, progressIndex, thisDataNodeId); + 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, + tConsensusGroupId, + thisDataNodeId); + 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)) { Review Comment: ditto -- 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]
