Pengzna commented on code in PR #12355: URL: https://github.com/apache/iotdb/pull/12355#discussion_r1617086716
########## 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(); Review Comment: this var is not useful anymore. I forgot to delete it when coding. fixed -- 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]
