ferenc-csaky commented on code in PR #261: URL: https://github.com/apache/flink-connector-aws/pull/261#discussion_r4495870252
########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/util/DynamoDbStreamsAsyncClientUtil.java: ########## @@ -0,0 +1,104 @@ +/* + * 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.connector.dynamodb.source.util; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.aws.config.AWSConfigConstants; +import org.apache.flink.connector.aws.util.AWSClientUtil; +import org.apache.flink.connector.aws.util.AWSGeneralUtil; +import org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants; + +import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; +import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption; +import software.amazon.awssdk.http.Protocol; +import software.amazon.awssdk.http.async.SdkAsyncHttpClient; +import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient; +import software.amazon.awssdk.retries.api.RetryStrategy; +import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsAsyncClient; +import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsAsyncClientBuilder; +import software.amazon.awssdk.utils.AttributeMap; + +import java.net.URI; +import java.util.Properties; + +/** + * Creates the {@link DynamoDbStreamsAsyncClient} that the source readers use for non-blocking + * GetRecords calls. The client has the same settings as the {@code DynamoDbStreamsClient} that + * {@link AWSClientUtil#createAwsSyncClient} creates for blocking calls. + */ +@Internal +public final class DynamoDbStreamsAsyncClientUtil { + + /** Same suffix that {@link AWSClientUtil} adds to the default user agent prefix. */ + private static final String V2_USER_AGENT_SUFFIX = " V2"; + + private DynamoDbStreamsAsyncClientUtil() {} + + /** + * Creates a Netty HTTP client with the same settings as the HTTP client for blocking calls. + * + * <p>{@link AWSGeneralUtil#createAsyncHttpClient} uses HTTP/2 by default. DynamoDB Streams + * endpoints use only HTTP/1.1, and an HTTP/2 client fails with "First received frame was not + * SETTINGS". A protocol set on the builder has precedence over the defaults of the helper, so + * the client always uses HTTP/1.1. + */ + public static SdkAsyncHttpClient createAsyncHttpClient() { + return AWSGeneralUtil.createAsyncHttpClient( + AttributeMap.empty(), NettyNioAsyncHttpClient.builder().protocol(Protocol.HTTP1_1)); Review Comment: Nothing tests this wiring against an endpoint. `DynamoDbStreamsAsyncClientUtilTest` checks the region and the endpoint override, and the reader and proxy tests use fake async clients. Per the description, an HTTP/2 client fails every call with "First received frame was not SETTINGS". So if this were switched to `AWSGeneralUtil.createAsyncHttpClient(Properties)`, which defaults to HTTP/2, every job on the default path would break and the unit tests would still pass. Could we add a test that sends one `getRecordsAsync` through the real Netty client to a local HTTP/1.1 stub? A source ITCase against the DynamoDB container, like the sink ones, would cover it too. ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/config/DynamodbStreamsSourceConfigConstants.java: ########## @@ -102,6 +102,23 @@ public enum InitialPosition { .withDescription( "The default idle time between non-empty polls for DynamoDB Streams GetRecords API"); + public static final ConfigOption<Boolean> DYNAMODB_STREAMS_GET_RECORDS_BLOCKING = + ConfigOptions.key("flink.dynamodbstreams.getrecords.blocking") Review Comment: This changes how every DynamoDB Streams source reads by default: one concurrent GetRecords call per shard over HTTP/1.1, so up to one connection per shard. This option is the way back to the old behaviour, but the docs don't mention it. Could we add it to the Configuration section of `docs/content/docs/connectors/datastream/dynamodb.md` and the `content.zh` copy, with a sentence on what the default now does? ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/util/DynamoDbStreamsAsyncClientUtil.java: ########## @@ -0,0 +1,104 @@ +/* + * 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.connector.dynamodb.source.util; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.aws.config.AWSConfigConstants; +import org.apache.flink.connector.aws.util.AWSClientUtil; +import org.apache.flink.connector.aws.util.AWSGeneralUtil; +import org.apache.flink.connector.dynamodb.source.config.DynamodbStreamsSourceConfigConstants; + +import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; +import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption; +import software.amazon.awssdk.http.Protocol; +import software.amazon.awssdk.http.async.SdkAsyncHttpClient; +import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient; +import software.amazon.awssdk.retries.api.RetryStrategy; +import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsAsyncClient; +import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsAsyncClientBuilder; +import software.amazon.awssdk.utils.AttributeMap; + +import java.net.URI; +import java.util.Properties; + +/** + * Creates the {@link DynamoDbStreamsAsyncClient} that the source readers use for non-blocking + * GetRecords calls. The client has the same settings as the {@code DynamoDbStreamsClient} that + * {@link AWSClientUtil#createAwsSyncClient} creates for blocking calls. + */ +@Internal +public final class DynamoDbStreamsAsyncClientUtil { + + /** Same suffix that {@link AWSClientUtil} adds to the default user agent prefix. */ + private static final String V2_USER_AGENT_SUFFIX = " V2"; + + private DynamoDbStreamsAsyncClientUtil() {} + + /** + * Creates a Netty HTTP client with the same settings as the HTTP client for blocking calls. + * + * <p>{@link AWSGeneralUtil#createAsyncHttpClient} uses HTTP/2 by default. DynamoDB Streams + * endpoints use only HTTP/1.1, and an HTTP/2 client fails with "First received frame was not + * SETTINGS". A protocol set on the builder has precedence over the defaults of the helper, so + * the client always uses HTTP/1.1. + */ + public static SdkAsyncHttpClient createAsyncHttpClient() { + return AWSGeneralUtil.createAsyncHttpClient( + AttributeMap.empty(), NettyNioAsyncHttpClient.builder().protocol(Protocol.HTTP1_1)); + } + + /** + * Creates the client. The client gets the region, credentials, endpoint override, and user + * agent prefix from {@code clientProperties}, as {@link AWSClientUtil#createAwsSyncClient} + * does. This method does not use {@link AWSClientUtil#createAwsAsyncClient}, because that + * method replaces the client override configuration and drops the retry strategy. + */ + public static DynamoDbStreamsAsyncClient createAsyncClient( Review Comment: This repeats `AWSClientUtil`'s client setup: the `V2` user-agent suffix, the user-agent prefix lookup, the endpoint override, and the credentials and region. As far as I can tell it's only needed because `AWSClientUtil.createAwsAsyncClient` builds its own `ClientOverrideConfiguration.builder()` and so can't take a retry strategy, while `createAwsSyncClient` has an overload that takes the builder. An async overload in aws-base would let this method go, and the HTTP client could be built inline in `DynamoDbStreamsSource`. Fine as a follow-up ticket if you'd rather not touch aws-base in this PR. ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/reader/PollingDynamoDbStreamsShardSplitReader.java: ########## @@ -144,14 +200,220 @@ public RecordsWithSplitIds<Record> fetch() throws IOException { splitContext.splitState.getStreamArn(), splitContext.splitState.getShardId(), splitContext.splitState.getNextStartingPosition()); + assignedSplits.add(splitContext); + return processGetRecordsResponse(splitContext, getRecordsResponse, currentTime); + } + + private RecordsWithSplitIds<Record> handleShardEnd( + DynamoDbStreamsShardSplitWithContext splitContext, long currentTime) { + if (!splitContext.hasAttemptedChildShardDiscovery) { + splitContext.hasAttemptedChildShardDiscovery = true; + splitContext.childShardDiscoveryAttempts = 0; + } + + if (splitContext.childShardDiscoveryAttempts < MAX_RETRY_ATTEMPTS_FOR_CHILD_SHARDS) { + long nextChildShardDiscoveryEligibleTime = + getNextEligibleTimeForChildDiscovery(splitContext); + if (currentTime >= nextChildShardDiscoveryEligibleTime) { + ListShardsResult listShardsResult = + dynamodbStreams.listShardsWithFilter( + splitContext.splitState.getStreamArn(), + childShardFilter(splitContext)); + RecordsWithSplitIds<Record> finishedSplit = + processChildShardDiscoveryResult( + splitContext, listShardsResult, currentTime); + if (finishedSplit != null) { + return finishedSplit; + } + } + assignedSplits.add(splitContext); + return INCOMPLETE_SHARD_EMPTY_RECORDS; + } else { + return finishedSplit(splitContext); + } + } + + private void sleep(long milliseconds) throws IOException { + try { + Thread.sleep(milliseconds); + } catch (InterruptedException e) { + throw new IOException("Split reader was interrupted while sleeping", e); + } + } + + // --------------------------------------------------------------------------------------- + // Non-blocking calls + // --------------------------------------------------------------------------------------- + + /** + * Starts a call for every shard that is ready and has no call in progress. Then collects one + * completed call. If no call is complete, waits until a call completes, a shard is ready, or + * {@link #wakeUp()} is called, for at most {@code NON_BLOCKING_FETCH_MAX_WAIT}. + */ + private RecordsWithSplitIds<Record> fetchNonBlocking() throws IOException { + long currentTime = System.currentTimeMillis(); + long nextEligibleTime = Long.MAX_VALUE; + + Iterator<DynamoDbStreamsShardSplitWithContext> splitIterator = assignedSplits.iterator(); + while (splitIterator.hasNext()) { + DynamoDbStreamsShardSplitWithContext splitContext = splitIterator.next(); + if (splitContext.callInProgress + || pausedSplitIds.contains(splitContext.splitState.getSplitId())) { + continue; + } + + long eligibleTime; + if (splitContext.splitState.isShardEndReached()) { + if (!splitContext.hasAttemptedChildShardDiscovery) { + splitContext.hasAttemptedChildShardDiscovery = true; + splitContext.childShardDiscoveryAttempts = 0; + } + if (splitContext.childShardDiscoveryAttempts + >= MAX_RETRY_ATTEMPTS_FOR_CHILD_SHARDS) { + splitIterator.remove(); + return finishedSplit(splitContext); + } + eligibleTime = getNextEligibleTimeForChildDiscovery(splitContext); + if (currentTime >= eligibleTime) { + startChildShardDiscovery(splitContext, currentTime); + continue; + } + } else { + eligibleTime = getNextEligibleTime(splitContext); + if (currentTime >= eligibleTime) { + startGetRecords(splitContext, currentTime); + continue; + } + } + nextEligibleTime = Math.min(nextEligibleTime, eligibleTime); + } + + CompletedCall completedCall = + pollResponseQueue( + Math.min( + NON_BLOCKING_FETCH_MAX_WAIT.toMillis(), + nextEligibleTime - currentTime)); + if (completedCall == null || completedCall == WAKE_UP) { + return INCOMPLETE_SHARD_EMPTY_RECORDS; + } + return handleCompletedCall(completedCall); + } + + private void startGetRecords( + DynamoDbStreamsShardSplitWithContext splitContext, long startTime) { + LOG.debug( + "Starting GetRecords for split: {}, wasEmptyPoll: {}", + splitContext.splitState.getSplitId(), + splitContext.wasLastPollEmpty); + splitContext.callInProgress = true; + dynamodbStreams + .getRecordsAsync( + splitContext.splitState.getStreamArn(), + splitContext.splitState.getShardId(), + splitContext.splitState.getNextStartingPosition()) + .whenComplete( + (response, error) -> + responseQueue.add( + new CompletedCall( + splitContext, + startTime, + false, + response, + null, + error))); + } + + private void startChildShardDiscovery( + DynamoDbStreamsShardSplitWithContext splitContext, long startTime) { + LOG.debug( + "Starting child shard discovery for split: {}, attempt: {}", + splitContext.splitState.getSplitId(), + splitContext.childShardDiscoveryAttempts); + splitContext.callInProgress = true; + dynamodbStreams + .listShardsWithFilterAsync( + splitContext.splitState.getStreamArn(), childShardFilter(splitContext)) + .whenComplete( + (result, error) -> + responseQueue.add( + new CompletedCall( + splitContext, + startTime, + true, + null, + result, + error))); + } + + @Nullable + private CompletedCall pollResponseQueue(long waitMillis) throws IOException { + try { + return responseQueue.poll(waitMillis, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Split reader was interrupted while waiting for records", e); + } + } + + private RecordsWithSplitIds<Record> handleCompletedCall(CompletedCall completedCall) + throws IOException { + DynamoDbStreamsShardSplitWithContext splitContext = completedCall.splitContext; + if (completedCall.error != null) { + // The call failed after all SDK retries. Fail the fetch, as a blocking call does. + throw rethrow(completedCall.error); + } + if (pausedSplitIds.contains(splitContext.splitState.getSplitId())) { + // The call stays in progress until the split resumes, so that the reader does not + // start a second call for the split. + heldCalls.put(splitContext.splitState.getSplitId(), completedCall); + return INCOMPLETE_SHARD_EMPTY_RECORDS; + } + splitContext.callInProgress = false; + + if (!completedCall.childShardDiscovery) { + return processGetRecordsResponse( + splitContext, completedCall.getRecordsResponse, completedCall.startTime); + } + RecordsWithSplitIds<Record> finishedSplit = + processChildShardDiscoveryResult( + splitContext, completedCall.listShardsResult, completedCall.startTime); + if (finishedSplit == null) { + return INCOMPLETE_SHARD_EMPTY_RECORDS; + } + assignedSplits.remove(splitContext); + return finishedSplit; + } + + private static IOException rethrow(Throwable error) { Review Comment: nit: `rethrow()` here and `unwrap()` in `DynamoDbStreamsProxy` (L519) strip `CompletionException` by hand. Flink's `ExceptionUtils` has `stripCompletionException`, and `rethrowIOException` does the "throw unchecked as is, wrap the rest in `IOException`" part. `ExceptionUtils.findThrowable`, which `AWSExceptionClassifierUtil` already uses, could replace the `instanceof` checks in `recoverGetRecordsAsync` and `getShardIteratorAsync`. ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/reader/PollingDynamoDbStreamsShardSplitReader.java: ########## @@ -144,14 +200,220 @@ public RecordsWithSplitIds<Record> fetch() throws IOException { splitContext.splitState.getStreamArn(), splitContext.splitState.getShardId(), splitContext.splitState.getNextStartingPosition()); + assignedSplits.add(splitContext); + return processGetRecordsResponse(splitContext, getRecordsResponse, currentTime); + } + + private RecordsWithSplitIds<Record> handleShardEnd( + DynamoDbStreamsShardSplitWithContext splitContext, long currentTime) { + if (!splitContext.hasAttemptedChildShardDiscovery) { + splitContext.hasAttemptedChildShardDiscovery = true; + splitContext.childShardDiscoveryAttempts = 0; + } + + if (splitContext.childShardDiscoveryAttempts < MAX_RETRY_ATTEMPTS_FOR_CHILD_SHARDS) { + long nextChildShardDiscoveryEligibleTime = + getNextEligibleTimeForChildDiscovery(splitContext); + if (currentTime >= nextChildShardDiscoveryEligibleTime) { + ListShardsResult listShardsResult = + dynamodbStreams.listShardsWithFilter( + splitContext.splitState.getStreamArn(), + childShardFilter(splitContext)); + RecordsWithSplitIds<Record> finishedSplit = + processChildShardDiscoveryResult( + splitContext, listShardsResult, currentTime); + if (finishedSplit != null) { + return finishedSplit; + } + } + assignedSplits.add(splitContext); + return INCOMPLETE_SHARD_EMPTY_RECORDS; + } else { + return finishedSplit(splitContext); + } + } + + private void sleep(long milliseconds) throws IOException { + try { + Thread.sleep(milliseconds); + } catch (InterruptedException e) { + throw new IOException("Split reader was interrupted while sleeping", e); + } + } + + // --------------------------------------------------------------------------------------- + // Non-blocking calls + // --------------------------------------------------------------------------------------- + + /** + * Starts a call for every shard that is ready and has no call in progress. Then collects one + * completed call. If no call is complete, waits until a call completes, a shard is ready, or + * {@link #wakeUp()} is called, for at most {@code NON_BLOCKING_FETCH_MAX_WAIT}. + */ + private RecordsWithSplitIds<Record> fetchNonBlocking() throws IOException { + long currentTime = System.currentTimeMillis(); + long nextEligibleTime = Long.MAX_VALUE; + + Iterator<DynamoDbStreamsShardSplitWithContext> splitIterator = assignedSplits.iterator(); + while (splitIterator.hasNext()) { + DynamoDbStreamsShardSplitWithContext splitContext = splitIterator.next(); + if (splitContext.callInProgress + || pausedSplitIds.contains(splitContext.splitState.getSplitId())) { + continue; + } + + long eligibleTime; + if (splitContext.splitState.isShardEndReached()) { + if (!splitContext.hasAttemptedChildShardDiscovery) { Review Comment: nit: this repeats the shard-end check from `handleShardEnd` (L209-214), including `childShardDiscoveryAttempts = 0`. That reset never changes the value: the counter starts at 0 and only goes up after shard end (L465). Could both paths use one helper, and `hasAttemptedChildShardDiscovery` go away? ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxy.java: ########## @@ -276,4 +477,49 @@ private GetRecordsResponse getRecords(String shardIterator) { return dynamoDbStreamsClient.getRecords( GetRecordsRequest.builder().shardIterator(shardIterator).build()); } + + private CompletableFuture<GetRecordsResponse> getRecordsAsync( + DynamoDbStreamsAsyncClient asyncClient, @Nullable String shardIterator) { + if (Objects.isNull(shardIterator)) { + return CompletableFuture.completedFuture(EMPTY_GET_RECORDS_RESPONSE); + } + return asyncClient.getRecords( + GetRecordsRequest.builder().shardIterator(shardIterator).build()); + } + + private GetRecordsResponse storeNextShardIterator( + String shardId, GetRecordsResponse getRecordsResponse) { + if (getRecordsResponse.nextShardIterator() != null) { + shardIdToIteratorStore.put(shardId, getRecordsResponse.nextShardIterator()); + } + return getRecordsResponse; + } + + private DynamoDbStreamsAsyncClient getAsyncClient() { + if (dynamoDbStreamsAsyncClient == null) { + throw new IllegalStateException( + "DynamoDbStreamsProxy has no DynamoDbStreamsAsyncClient. " + + "Non-blocking calls are not available."); + } + return dynamoDbStreamsAsyncClient; + } + + private static void logIfStreamIsEnabling( Review Comment: nit: only the async path calls `logIfStreamIsEnabling`. The two sync `describeStream` methods (L338 and L362) still have their own copies of the warning; could they call it too? The filtered `DescribeStreamRequest` is also built in two places (L284 and L352), which a `describeStreamRequest(streamArn, shardFilter)` helper would cover. ########## flink-connector-aws/flink-connector-dynamodb/src/test/java/org/apache/flink/connector/dynamodb/source/reader/PollingDynamoDbStreamsShardSplitReaderTest.java: ########## @@ -69,14 +70,56 @@ public void init() { new DynamoDbStreamsShardMetrics( TestUtil.getTestSplit(TEST_SHARD_ID), metricListener.getMetricGroup())); splitReader = - new PollingDynamoDbStreamsShardSplitReader( + createSplitReader( testStreamProxy, NON_EMPTY_POLLING_DELAY_MILLIS, EMPTY_POLLING_DELAY_MILLIS, new ConcurrentHashMap<>(), shardMetricGroupMap); } + /** Returns true to run the tests with blocking GetRecords calls. */ + protected boolean blockingGetRecords() { + return false; Review Comment: nit: with this returning `false`, the inherited `testWakeUpIsNoOp` (L299) now runs in non-blocking mode, where `wakeUp()` enqueues `WAKE_UP`, so it isn't a no-op any more. Could that test move to the blocking subclass, with a non-blocking test that checks `wakeUp()` makes a waiting `fetch()` return early? ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/reader/DynamoDbStreamsSourceReader.java: ########## @@ -68,7 +75,30 @@ public DynamoDbStreamsSourceReader( SourceReaderContext context, Map<String, List<Shard>> childShardIdMap, Map<String, DynamoDbStreamsShardMetrics> shardMetricGroupMap) { + this( + splitFetcherManager, + recordEmitter, + config, + context, + childShardIdMap, + shardMetricGroupMap, + null); Review Comment: nit: this constructor (with a null proxy) is only used by tests, and so is the 5-arg `PollingDynamoDbStreamsShardSplitReader` constructor. The private 2-arg `createDynamoDbStreamsProxy` has one caller. The 5-arg split reader constructor means non-blocking, but the 2-arg `DynamoDbStreamsProxy` constructor builds no async client, so the two short forms don't go together. Passing `null`/`false`/`true` at the call sites and dropping the three overloads would save about 35 lines. ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSource.java: ########## @@ -160,19 +164,30 @@ public SourceReader<T, DynamoDbStreamsShardSplit> createReader( final Duration getRecordsIdlePollingTimeBetweenEmptyPolls = sourceConfig.get(DYNAMODB_STREAMS_GET_RECORDS_IDLE_TIME_BETWEEN_EMPTY_POLLS); + // If true, the split reader calls GetRecords on one shard at a time and waits for the + // response. Default - false + final boolean blockingGetRecords = sourceConfig.get(DYNAMODB_STREAMS_GET_RECORDS_BLOCKING); + Map<String, List<Shard>> childShardMap = new ConcurrentHashMap<>(); - // We create a new stream proxy for each split reader since they have their own independent - // lifecycle. + // All split readers of this source reader share one stream proxy. The fetcher manager + // creates a new split reader each time it starts a fetcher (for example, after all splits + // of the previous fetcher finished), and a proxy per split reader would create a new AWS + // SDK client, HTTP client and, for non-blocking calls, Netty event loop each time. The + // source reader closes the proxy. + DynamoDbStreamsProxy dynamoDbStreamsProxy = + createDynamoDbStreamsProxy( + sourceConfig, + SdkDefaultRetryStrategy.defaultRetryStrategy(), Review Comment: All of a reader's calls go through clients built with this one strategy instance, so they share its retry token bucket. In SDK 2.56 the bucket has 500 tokens, and each retry of a non-throttling error (5xx, connection reset) costs 5 in legacy mode and 5 or 14 in standard mode. Throttling costs nothing in legacy mode (see `DefaultRetryStrategy.Legacy` and `.Standard`). With about 100 calls in flight, one short burst of 5xx responses spends the whole bucket on the first retries. After that the remaining calls fail without a retry, `handleCompletedCall` rethrows, and the job fails over. In blocking mode the same burst hits one call at a time, and the retries usually get through it. Could the reader's strategy turn off the circuit breaker? For example `SdkDefaultRetryStrategy.legacyRetryStrategyBuilder().circuitBreakerEnabled(false).build()`; the standard builder has the same switch. The in-flight cap from the comment on `PollingDynamoDbStreamsShardSplitReader` would also limit how many calls one burst can hit. ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/reader/PollingDynamoDbStreamsShardSplitReader.java: ########## @@ -144,14 +200,220 @@ public RecordsWithSplitIds<Record> fetch() throws IOException { splitContext.splitState.getStreamArn(), splitContext.splitState.getShardId(), splitContext.splitState.getNextStartingPosition()); + assignedSplits.add(splitContext); + return processGetRecordsResponse(splitContext, getRecordsResponse, currentTime); + } + + private RecordsWithSplitIds<Record> handleShardEnd( + DynamoDbStreamsShardSplitWithContext splitContext, long currentTime) { + if (!splitContext.hasAttemptedChildShardDiscovery) { + splitContext.hasAttemptedChildShardDiscovery = true; + splitContext.childShardDiscoveryAttempts = 0; + } + + if (splitContext.childShardDiscoveryAttempts < MAX_RETRY_ATTEMPTS_FOR_CHILD_SHARDS) { + long nextChildShardDiscoveryEligibleTime = + getNextEligibleTimeForChildDiscovery(splitContext); + if (currentTime >= nextChildShardDiscoveryEligibleTime) { + ListShardsResult listShardsResult = + dynamodbStreams.listShardsWithFilter( + splitContext.splitState.getStreamArn(), + childShardFilter(splitContext)); + RecordsWithSplitIds<Record> finishedSplit = + processChildShardDiscoveryResult( + splitContext, listShardsResult, currentTime); + if (finishedSplit != null) { + return finishedSplit; + } + } + assignedSplits.add(splitContext); + return INCOMPLETE_SHARD_EMPTY_RECORDS; + } else { + return finishedSplit(splitContext); + } + } + + private void sleep(long milliseconds) throws IOException { + try { + Thread.sleep(milliseconds); + } catch (InterruptedException e) { + throw new IOException("Split reader was interrupted while sleeping", e); + } + } + + // --------------------------------------------------------------------------------------- + // Non-blocking calls + // --------------------------------------------------------------------------------------- + + /** + * Starts a call for every shard that is ready and has no call in progress. Then collects one + * completed call. If no call is complete, waits until a call completes, a shard is ready, or + * {@link #wakeUp()} is called, for at most {@code NON_BLOCKING_FETCH_MAX_WAIT}. + */ + private RecordsWithSplitIds<Record> fetchNonBlocking() throws IOException { + long currentTime = System.currentTimeMillis(); + long nextEligibleTime = Long.MAX_VALUE; + + Iterator<DynamoDbStreamsShardSplitWithContext> splitIterator = assignedSplits.iterator(); + while (splitIterator.hasNext()) { + DynamoDbStreamsShardSplitWithContext splitContext = splitIterator.next(); + if (splitContext.callInProgress + || pausedSplitIds.contains(splitContext.splitState.getSplitId())) { + continue; + } + + long eligibleTime; + if (splitContext.splitState.isShardEndReached()) { + if (!splitContext.hasAttemptedChildShardDiscovery) { + splitContext.hasAttemptedChildShardDiscovery = true; + splitContext.childShardDiscoveryAttempts = 0; + } + if (splitContext.childShardDiscoveryAttempts + >= MAX_RETRY_ATTEMPTS_FOR_CHILD_SHARDS) { + splitIterator.remove(); + return finishedSplit(splitContext); + } + eligibleTime = getNextEligibleTimeForChildDiscovery(splitContext); + if (currentTime >= eligibleTime) { + startChildShardDiscovery(splitContext, currentTime); + continue; + } + } else { + eligibleTime = getNextEligibleTime(splitContext); + if (currentTime >= eligibleTime) { + startGetRecords(splitContext, currentTime); Review Comment: This starts a GetRecords call for every shard that's due, but each `fetch()` returns only one completed call, and `callInProgress` is cleared only when that call is collected (L371). When the reader is backpressured or catching up, the fetcher thread waits in `elementsQueue.put` while the other calls complete, so every assigned shard can end up with a full response sitting in `responseQueue` (or in `heldCalls` for a paused split). Nothing limits the total. Flink's element queue doesn't bound it either, since the responses are in the split reader's own queue. A GetRecords response carries up to 1 MB or 1,000 records, and takes more heap than that once unmarshalled. About 100 shards per reader, as in the load test, is fine. A large table on low parallelism is not: during catch-up, when responses are full, 250 shards per reader means 250 MB or more of responses on the heap, where the blocking reader held one. Could we cap the calls in flight per reader? Count up in `startGetRecords`/`startChildShardDiscovery`, count down in `handleCompletedCall`, and stop starting new calls in `fetchNonBlocking` once the cap is reached. An option such as `flink.dynamodbstreams.getrecords.max-in-flight` would let users tune it. If that doesn't fit in this PR, I'd keep blocking as the default for a release. ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxy.java: ########## @@ -276,4 +477,49 @@ private GetRecordsResponse getRecords(String shardIterator) { return dynamoDbStreamsClient.getRecords( GetRecordsRequest.builder().shardIterator(shardIterator).build()); } + + private CompletableFuture<GetRecordsResponse> getRecordsAsync( + DynamoDbStreamsAsyncClient asyncClient, @Nullable String shardIterator) { + if (Objects.isNull(shardIterator)) { + return CompletableFuture.completedFuture(EMPTY_GET_RECORDS_RESPONSE); + } + return asyncClient.getRecords( + GetRecordsRequest.builder().shardIterator(shardIterator).build()); + } + + private GetRecordsResponse storeNextShardIterator( + String shardId, GetRecordsResponse getRecordsResponse) { + if (getRecordsResponse.nextShardIterator() != null) { Review Comment: Nothing removes entries from `shardIdToIteratorStore`. That isn't new, but each split reader used to own its proxy, so the map went away whenever a fetcher shut down. Now the proxy lives as long as the source reader, and the map keeps an iterator for every shard the reader has ever read. Stream shards roll over roughly every 4 hours, so a reader with 100 shards adds about 600 entries a day. Could we `remove(shardId)` here when `nextShardIterator()` is null, and do the same in the blocking `getRecords` paths? ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/proxy/DynamoDbStreamsProxy.java: ########## @@ -160,10 +190,138 @@ public GetRecordsResponse getRecords( } } + /** + * Non-blocking version of {@link #getRecords(String, String, StartingPosition)}. Each step + * matches a step of the blocking method: the shard iterator store, the retry after + * ExpiredIteratorException, the read from TRIM_HORIZON after TrimmedDataAccessException, and + * the empty response after ResourceNotFoundException. + */ + @Override + public CompletableFuture<GetRecordsResponse> getRecordsAsync( + String streamArn, String shardId, StartingPosition startingPosition) { + DynamoDbStreamsAsyncClient asyncClient = getAsyncClient(); + + String storedShardIterator = shardIdToIteratorStore.get(shardId); + CompletableFuture<String> shardIteratorFuture = + storedShardIterator != null + ? CompletableFuture.completedFuture(storedShardIterator) + : getShardIteratorAsync(asyncClient, streamArn, shardId, startingPosition) + .thenApply( + shardIterator -> { + if (shardIterator != null) { + shardIdToIteratorStore.put(shardId, shardIterator); + } + return shardIterator; + }); + + return shardIteratorFuture.thenCompose( + shardIterator -> { + if (shardIterator == null) { + return CompletableFuture.completedFuture(EMPTY_GET_RECORDS_RESPONSE); + } + return getRecordsAsync(asyncClient, shardIterator) + .thenApply(response -> storeNextShardIterator(shardId, response)) + // Same as exceptionallyCompose, which needs Java 12. + .handle( + (response, error) -> + error == null + ? CompletableFuture.completedFuture(response) + : recoverGetRecordsAsync( + asyncClient, + streamArn, + shardId, + startingPosition, + unwrap(error))) + .thenCompose(Function.identity()); + }); + } + + private CompletableFuture<GetRecordsResponse> recoverGetRecordsAsync( + DynamoDbStreamsAsyncClient asyncClient, + String streamArn, + String shardId, + StartingPosition startingPosition, + Throwable error) { + if (error instanceof ExpiredIteratorException) { + LOG.info( + "Received ExpiredIteratorException from GetRecords. " + + "Calling GetShardIterator for shard: {} with position: {}", + shardId, + startingPosition); + // Eagerly retry getRecords() if the iterator is expired + return getShardIteratorAsync(asyncClient, streamArn, shardId, startingPosition) + .thenCompose(shardIterator -> getRecordsAsync(asyncClient, shardIterator)) + .thenApply(response -> storeNextShardIterator(shardId, response)); + } + if (error instanceof TrimmedDataAccessException) { + // TrimmedDataAccessException means that the record pointed by shard iterator has + // expired. We should read the shard back from TRIM_HORIZON + LOG.warn( + "Received TrimmedDataAccessException from GetRecords. " + + "Calling GetShardIterator for shard: {} with TRIM_HORIZON", + shardId); + return getShardIteratorAsync( + asyncClient, streamArn, shardId, StartingPosition.fromStart()) + .thenCompose(shardIterator -> getRecordsAsync(asyncClient, shardIterator)) + .thenApply(response -> storeNextShardIterator(shardId, response)); + } + if (error instanceof ResourceNotFoundException) { + LOG.warn( + "Received ResourceNotFoundException from GetRecords for shard: {}. " + + "This might indicate that there is restore happening from stale snapshot or data loss from backpressure", + shardId); + return CompletableFuture.completedFuture(EMPTY_GET_RECORDS_RESPONSE); + } + return CompletableFuture.failedFuture(error); + } + + /** Non-blocking version of {@link #listShardsWithFilter(String, ShardFilter)}. */ + @Override + public CompletableFuture<ListShardsResult> listShardsWithFilterAsync( + String streamArn, ShardFilter shardFilter) { + DynamoDbStreamsAsyncClient asyncClient = getAsyncClient(); + LOG.debug("Child shards with filter called, for shardId: {}", shardFilter.shardId()); + final DescribeStreamRequest describeStreamRequest = + DescribeStreamRequest.builder() + .streamArn(streamArn) + .shardFilter(shardFilter) + .build(); + return asyncClient + .describeStream(describeStreamRequest) + .thenApply( + describeStreamResponse -> { + logIfStreamIsEnabling(streamArn, describeStreamResponse); + ListShardsResult listShardsResult = new ListShardsResult(); + listShardsResult.addShards( + describeStreamResponse.streamDescription().shards()); + listShardsResult.setStreamStatus( + describeStreamResponse.streamDescription().streamStatus()); + return listShardsResult; + }) + .exceptionally( + error -> { + LOG.warn( + "DescribeStream with Filter API threw an exception", + unwrap(error)); + return new ListShardsResult(); + }) + .thenApply( + listShardsResult -> { + LOG.info("Child shards returned for shardId: {}", listShardsResult); + return listShardsResult; + }); + } + @Override public void close() throws IOException { Review Comment: If `dynamoDbStreamsClient.close()` or `httpClient.close()` throws, the async client and the Netty client with its event-loop threads stay open. `AWSGeneralUtil.closeResources(dynamoDbStreamsClient, httpClient, dynamoDbStreamsAsyncClient, asyncHttpClient)` skips nulls and keeps closing after a failure, and the sinks already use it. ########## flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/DynamoDbStreamsSource.java: ########## @@ -259,7 +284,15 @@ private DynamoDbStreamsProxy createDynamoDbStreamsProxy( DynamodbStreamsSourceConfigConstants .BASE_DDB_STREAMS_USER_AGENT_PREFIX_FORMAT, DynamodbStreamsSourceConfigConstants.DDB_STREAMS_CLIENT_USER_AGENT_PREFIX); - return new DynamoDbStreamsProxy(dynamoDbStreamsClient, httpClient); + if (!withAsyncClient) { Review Comment: In non-blocking mode the proxy still gets a sync `DynamoDbStreamsClient`, an Apache HTTP client and its own credentials provider (a second STS client with ASSUME_ROLE or WEB_IDENTITY), but the split reader only calls the async methods. And since the proxy is now created in `createReader`, a reader that never gets a split, for example with parallelism above the shard count, keeps both clients and a Netty event loop open for the whole job. Could non-blocking mode build an async-only proxy, or create the proxy when the first split reader needs it? -- 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]
