dannycranmer commented on a change in pull request #13102: URL: https://github.com/apache/flink/pull/13102#discussion_r490179387
########## 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()); + SequenceNumber sequenceNumber = recordConsumer.accept(recordBatch); + nextStartingPosition = StartingPosition.continueFromSequenceNumber(sequenceNumber); + }; + + RecordPublisherRunResult result = runWithBackoff(eventConsumer); + + LOG.info("Subscription expired {}::{}, with status {}", + subscribedShard.getStreamName(), + subscribedShard.getShard().getShardId(), + result); + + return result; + } + + /** + * Runs the record publisher, will sleep for configuration computed jitter period in the case of certain exceptions. + * Unrecoverable exceptions are thrown to terminate the application. + * + * @param eventConsumer the consumer to pass events to + * @return {@code COMPLETE} if the shard is complete and this shard consumer should exit + * @throws InterruptedException + */ + private RecordPublisherRunResult runWithBackoff( + final Consumer<SubscribeToShardEvent> eventConsumer) throws InterruptedException { + FanOutShardSubscriber fanOutShardSubscriber = new FanOutShardSubscriber( + consumerArn, + subscribedShard.getShard().getShardId(), + kinesisProxy); + boolean complete; + + try { + complete = fanOutShardSubscriber.subscribeToShardAndConsumeRecords( + toSdkV2StartingPosition(nextStartingPosition), eventConsumer); + attempt = 0; + } catch (FanOutSubscriberException ex) { + // We have received an error from the network layer + // This can be due to limits being exceeded, network timeouts, etc + // We should backoff, reacquire a subscription and try again + if (ex.getCause() instanceof ResourceNotFoundException) { + LOG.warn("Received ResourceNotFoundException. Either the shard does not exist, or the stream subscriber has been deregistered." + + "Marking this shard as complete {} ({})", subscribedShard.getShard().getShardId(), consumerArn); + + return COMPLETE; + } + + if (attempt == configuration.getSubscribeToShardMaxRetries()) { + throw new RuntimeException("Maximum reties exceeded for SubscribeToShard. " + + "Failed " + configuration.getSubscribeToShardMaxRetries() + " times."); + } + + backoff(ex); + return INCOMPLETE; + } + + return complete ? COMPLETE : INCOMPLETE; + } + + private void backoff(final Throwable ex) throws InterruptedException { + long backoffMillis = backoff.calculateFullJitterBackoff( + configuration.getSubscribeToShardBaseBackoffMillis(), + configuration.getSubscribeToShardMaxBackoffMillis(), + configuration.getSubscribeToShardExpConstant(), + ++attempt); Review comment: Yep, will do ---------------------------------------------------------------- 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
