ferenc-csaky commented on code in PR #258:
URL:
https://github.com/apache/flink-connector-aws/pull/258#discussion_r4388667134
##########
pom.xml:
##########
@@ -53,7 +53,7 @@ under the License.
</scm>
<properties>
- <aws.sdkv2.version>2.40.3</aws.sdkv2.version>
+ <aws.sdkv2.version>2.56.0</aws.sdkv2.version>
Review Comment:
`apache5-client` is now shaded into all three SQL jars and registered as an
`SdkHttpService`, but HttpClient 5 isn't bundled
(`org.apache.httpcomponents.client5`/`core5` don't match the include). The SDK
prefers Apache5, so clients built without an explicit `httpClient(...)` fail.
That covers STS in `AWSGeneralUtil#getAssumeRoleCredentialProvider` and web
identity (`WEB_IDENTITY_TOKEN`, or `AUTO` on EKS/IRSA):
```
java.lang.NoClassDefFoundError: org/apache/hc/core5/http/config/Lookup
```
Could you add this to the shade `<artifactSet>`, right after `</includes>`,
in the `pom.xml` of `flink-sql-connector-dynamodb`,
`flink-sql-connector-aws-kinesis-streams` and
`flink-sql-connector-aws-kinesis-firehose`, and drop `apache5-client` from
their NOTICE files?
```xml
<excludes>
<exclude>software.amazon.awssdk:apache5-client</exclude>
</excludes>
```
##########
flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/enumerator/tracker/SplitTracker.java:
##########
@@ -179,13 +197,75 @@ private void addSplitToMapping(DynamoDbStreamsShardSplit
split) {
.add(split.splitId());
}
+ /**
+ * Adds shards discovered under the AT_TIMESTAMP initial position. Each
shard that is not
+ * already tracked is added at AT_TIMESTAMP:
GetShardIterator(AT_TIMESTAMP) resolves the
+ * per-record position, so a shard that straddles the timestamp never
emits pre-timestamp
+ * records, and the position does not depend on the discovery path or
restore timing.
+ *
+ * <p>Ancestors of already-tracked shards are not added, to keep the
parent-before-child order.
+ * Usually they have already been read. An ancestor that was never tracked
(for example, one an
+ * earlier discovery did not return) is skipped too, so any of its records
at or after the
+ * timestamp are not read; a warning is logged when that happens.
+ */
+ private void addSplitsForAtTimestamp(Collection<Shard> shardsToAdd) {
+ Map<String, Shard> shardIdToShardMap =
+ shardsToAdd.stream().collect(Collectors.toMap(Shard::shardId,
shard -> shard));
+
+ Set<String> ancestorsOfTrackedSplits =
findAncestorsOfTrackedSplits(shardIdToShardMap);
+
+ List<Shard> eligibleShards = new ArrayList<>();
+ for (Shard shard : shardsToAdd) {
+ String shardId = shard.shardId();
+ if (!ancestorsOfTrackedSplits.contains(shardId)) {
+ eligibleShards.add(shard);
+ } else if (!knownSplits.containsKey(shardId)) {
+ LOG.warn(
Review Comment:
Nit: the skipped shard is never tracked, so this WARN repeats on every
discovery (60s by default) while the shard is listed, up to ~1,440 times over
24h. Could we warn once per shard?
##########
flink-connector-aws/flink-connector-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkWriterTest.java:
##########
@@ -50,7 +50,7 @@ public class KinesisFirehoseSinkWriterTest {
@BeforeEach
void setup() throws IOException {
TestSinkInitContext sinkInitContext = new TestSinkInitContext();
- Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake_aws_endpoint";);
+ Properties sinkProperties =
AWSServicesTestUtils.createConfig("https://fake-aws-endpoint";);
Review Comment:
Thanks, confirmed. This also affects users: with 2.56.0, an `aws.endpoint`
with an underscore in the host fails. Worth a release note on the JIRA.
--
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]
