dannycranmer commented on a change in pull request #13102: URL: https://github.com/apache/flink/pull/13102#discussion_r472120604
########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/publisher/fanout/FanOutRecordPublisher.java ########## @@ -0,0 +1,245 @@ +/* + * 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.flink.streaming.connectors.kinesis.internals.publisher.fanout; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordBatch; +import org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher; +import org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber.FanOutSubscriberException; +import org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber; +import org.apache.flink.streaming.connectors.kinesis.model.StartingPosition; +import org.apache.flink.streaming.connectors.kinesis.model.StreamShardHandle; +import org.apache.flink.streaming.connectors.kinesis.proxy.FullJitterBackoff; +import org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxyV2Interface; +import org.apache.flink.util.Preconditions; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.kinesis.model.EncryptionType; +import software.amazon.awssdk.services.kinesis.model.Record; +import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException; +import software.amazon.awssdk.services.kinesis.model.SubscribeToShardEvent; + +import javax.annotation.Nonnull; + +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.function.Consumer; + +import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.COMPLETE; +import static org.apache.flink.streaming.connectors.kinesis.internals.publisher.RecordPublisher.RecordPublisherRunResult.INCOMPLETE; +import static software.amazon.awssdk.services.kinesis.model.StartingPosition.builder; + +/** + * A {@link RecordPublisher} that will read and forward records from Kinesis using EFO, to the subscriber. + * Records are consumed via Enhanced Fan Out subscriptions using SubscribeToShard API. + */ +@Internal +public class FanOutRecordPublisher implements RecordPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(FanOutRecordPublisher.class); + + private final FullJitterBackoff backoff; + + private final String consumerArn; + + private final KinesisProxyV2Interface kinesisProxy; + + private final StreamShardHandle subscribedShard; + + private final FanOutRecordPublisherConfiguration configuration; + + /** The current attempt in the case of subsequent recoverable errors. */ + private int attempt = 0; + + private StartingPosition nextStartingPosition; + + /** + * Instantiate a new FanOutRecordPublisher. + * Consumes data from KDS using EFO SubscribeToShard over AWS SDK V2.x + * + * @param startingPosition the position in the shard to start consuming from + * @param consumerArn the consumer ARN of the stream consumer + * @param subscribedShard the shard to consumer from + * @param kinesisProxy the proxy used to talk to Kinesis services + * @param configuration the record publisher configuration + */ + public FanOutRecordPublisher( + final StartingPosition startingPosition, + final String consumerArn, + final StreamShardHandle subscribedShard, + final KinesisProxyV2Interface kinesisProxy, + final FanOutRecordPublisherConfiguration configuration, + final FullJitterBackoff backoff) { + this.nextStartingPosition = Preconditions.checkNotNull(startingPosition); + this.consumerArn = Preconditions.checkNotNull(consumerArn); + this.subscribedShard = Preconditions.checkNotNull(subscribedShard); + this.kinesisProxy = Preconditions.checkNotNull(kinesisProxy); + this.configuration = Preconditions.checkNotNull(configuration); + this.backoff = Preconditions.checkNotNull(backoff); + } + + @Override + public RecordPublisherRunResult run(final RecordBatchConsumer recordConsumer) throws InterruptedException { + LOG.info("Running fan out record publisher on {}::{} from {} - {}", + subscribedShard.getStreamName(), + subscribedShard.getShard().getShardId(), + nextStartingPosition.getShardIteratorType(), + nextStartingPosition.getStartingMarker()); + + Consumer<SubscribeToShardEvent> eventConsumer = event -> { + RecordBatch recordBatch = new RecordBatch(toSdkV1Records(event.records()), subscribedShard, event.millisBehindLatest()); Review comment: @xiaolong-sn I have verified that the `data` is passed by reference, it is not copied. So the overhead would be the GC on the new Object inflation. The rest is shuffling references. Once we update Polling to SDK V2 we can use KCL v2.x and get rid of the transform. The alternative now is to import KCL v2.x which would bloat the dependency chain and application size, or copy the deaggregation code which would add a maintenance burden. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
