gguptp commented on code in PR #261:
URL:
https://github.com/apache/flink-connector-aws/pull/261#discussion_r4505114835
##########
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:
added `flink.dynamodbstreams.getrecords.max-in-flight`, default 100. it
counts every call from start until the response is collected. when the limit is
hit, the splits that were kept waiting move to the front of the queue, so they
get the next free calls and the first shards can't starve the rest. held
responses of paused splits don't count, since with watermark alignment the
paused splits could otherwise take every call and block the splits that are
behind. so the bound is the cap plus one response per paused split. tests:
`testStartsNoMoreThanMaxCallsInFlight`, `testWaitingShardsGetFreeCallsFirst`,
`testHeldCallOfPausedSplitFreesItsCall`.
##########
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:
good catch. the split readers now use
`legacyRetryStrategyBuilder().circuitBreakerEnabled(false)`, which keeps the
legacy attempts and backoff and only turns off the bucket. the enumerator keeps
its adaptive strategy.
##########
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:
added a "GetRecords Calls" section under Configuration that covers both
options and what the default does now. the content.zh dynamodb page only has
the sink sections, no source section at all, so i left it out there. happy to
copy the whole source section over in a separate PR if you want.
##########
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:
done, the store now removes the shard when the next iterator is null, in the
async path and all three sync 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:
done, switched to `AWSGeneralUtil.closeResources`.
##########
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:
added `testAsyncClientCallsHttp11Endpoint`, now in
`DynamoDbStreamsSourceAsyncClientTest` since the util class is gone. it sends
one GetRecords through the real netty client to a local HTTP/1.1 server. i
checked that it fails with "First received frame was not SETTINGS" if the
client is switched to HTTP/2.
--
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]
