Amazon Timestream Connector gets stuck during checkpointing Hi! I'm using Amazon Timestream Connector for Apache Flink to write processed records into Timestream database. Writing records is implemented in this asyncWriteRecords method. It seems that when checkpointing starts, these sink operators get stuck waiting for something. It's hard to understand what happens, but it possibly happens in this snapshotState() method. Does anyone have any idea why an async sink might get stuck during checkpointing? Thanks so much 🙏

Hi, there is an issue in AWS SDK that causes deadlock in async http clients. We have seen this in other AWS connectors. This should be fixed in SDK starting from version 2.20.144 Can you check if updating AWS SDK helps?

Thanks so much for the help, Aleksandr! We have 2.21.5 for all AWS SDK dependencies except for

software.amazon.awssdk:aws-crt-client:2.17.204-PREVIEW

, because the Timestream sink connector has refused to work with new versions 😞 Should we then just fork the Timestream Sink connector and change it so that we can use a recent version of aws-crt-client 🤔 Or create a GitHub issue to ask for updating to most recent version.

I think it would be better to engage with Timestream team by opening an issue.

We have 2.21.5 for all AWS SDK dependencies except for

software.amazon.awssdk:aws-crt-client:2.17.204-PREVIEW

, because the Timestream sink connector has refused to work with new versions

I see that code in Github uses newer version: 2.20.101, are you using older revision? https://github.com/awslabs/amazon-timestream-tools/blob/mainline/integrations/flink_connector/flink-connector-timestream/pom.xml#L92

Thanks, I'll look into it again 👍

I see that method signature had changed somewhere between v2.17.x and v2.20.x. I suspect that timestream connector may need to be recompiled for it to work. I can’t test this at the moment, unfortunately.

Thanks Aleksandr! We were able to fix this by compiling the sink connector source (without any changes) and including the JAR in our Flink jar. Everything running smoothly now in production 🎉 Thanks again!

Great!

Small update on this issue. It turned out that simply updating AWS SDK version for the affected libraries in

build.gradle

did not fix the issue. We still ran into deadlocks in the Timestream sink connector. It just took some time for the deadlock to happen. What we're trying now is to upgrade BOM to ensure that a recent enough AWS SDK version is enforced in all dependencies. Looking good so far 🤞

Thank you for the update. Could you verify that your resulting jar contains SDK fix?

Thanks! TBH I'm not sure how to verify that, especially at runtime because I think aws managed flink also introduces some aws sdk dependencies at runtime... We did this in our build.gradle:

// Force AWS SDK update by adding a platform dependency
// <https://docs.aws.amazon.com/managed-flink/latest/java/troubleshooting-async-deadlock-update-java-apps.html>
flinkShadowJar(platform("software.amazon.awssdk:bom:2.20.144"))

We have a self-compiled JAR of flink-connector-timestream in our

libs

folder that we include like this:

flinkShadowJar files('libs/flink-connector-timestream-0.4-SNAPSHOT.jar')

If I do

./gradlew -q dependencies | grep aws

, all dependencies are either 2.20.144 or 2.21.5 (which both should contain the fix). flink-connector-timestream also contains this in the pom to force dependencies:

<dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>software.amazon.awssdk</groupId>
                <artifactId>bom</artifactId>
                <version>2.20.101</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

What we haven't tried yet is changing this to

2.20.144

before compiling the JAR. Not sure if that would help anything though.

This may be able to help. Regarding a way to confirm that fix is included - I usually check decompiled classes in resulting jar.

Update on this issue. We tried using

enforcedPlatform

to enforce the versions of AWS SDK in the runtime to 2.20.144. This did not help either, deadlock happened again. Luckily this time we were clever enough to let this run long enough in the test environment before deploying to production 😄

Hi, ServiceResourceTransformer shouldn’t play a role here. Would you be able to share your gradle build file? Just a part with connector + sdk + shading

Thanks Aleksandr! Here is part of our build.gradle:

plugins {
    id 'scala'
    id 'application'
    id("com.github.maiflai.scalatest") version "0.32"
    id 'com.github.johnrengelman.shadow' version '7.1.2'
}

ext {
    flinkVersion = '1.15.2'
    scalaBinaryVersion = '2.12'
    slf4jVersion = '1.7.32'
    log4jVersion = '2.17.1'
    awsSDKVersion = '2.20.144'
    circeVersion = '0.14.6'
    javaVersion = JavaVersion.VERSION_11
}

java {
    sourceCompatibility = javaVersion
    targetCompatibility = javaVersion
}

tasks.withType(JavaCompile) {
    options.encoding = 'UTF-8'
}
tasks.withType(ScalaCompile) {
    scalaCompileOptions.additionalParameters = [ 
        "-deprecation",
        "-feature",
        "-unchecked",
        "-language:implicitConversions",
        "-language:higherKinds",
        "-language:existentials",
        "-language:postfixOps",
        "-Ywarn-unused"
    ]
}

repositories {
    mavenCentral()
    maven {
        url "<https://repository.apache.org/content/repositories/snapshots>"
        mavenContent {
            snapshotsOnly()
        }
    }
}

// NOTE: We cannot use "compileOnly" or "shadow" configurations since then we could not run code
// in the IDE or with "gradle run". We also cannot exclude transitive dependencies from the
// shadowJar yet (see <https://github.com/johnrengelman/shadow/issues/159>).
// -> Explicitly define the // libraries we want to be included in the "flinkShadowJar" configuration!
configurations {
    flinkShadowJar

    // always exclude these (also from transitive dependencies) since they are provided by Flink
    flinkShadowJar.exclude group: 'org.apache.flink', module: 'force-shading'
    flinkShadowJar.exclude group: 'com.google.code.findbugs', module: 'jsr305'
    flinkShadowJar.exclude group: 'org.slf4j'
    flinkShadowJar.exclude group: 'org.apache.logging.log4j'
}

dependencies {
    // --------------------------------------------------------------
    // Compile-time dependencies that should NOT be part of the
    // shadow jar and are provided in the lib folder of Flink
    // --------------------------------------------------------------
    implementation "org.apache.flink:flink-json:${flinkVersion}"
    implementation "org.apache.flink:flink-clients:${flinkVersion}"
    implementation "org.scala-lang:scala-library:${scalaBinaryVersion}.17"
    implementation "com.amazonaws:aws-kinesisanalytics-runtime:1.2.0"

    // --------------------------------------------------------------
    // Dependencies that should be part of the shadow jar, e.g.
    // connectors. These must be in the flinkShadowJar configuration!
    // --------------------------------------------------------------
    flinkShadowJar "org.apache.flink:flink-connector-kinesis:${flinkVersion}"
    // Fixes AWSConfigConstants being empty
    flinkShadowJar "org.apache.flink:flink-connector-aws-base:${flinkVersion}"

    // Force AWS SDK update by adding a platform dependency
    // <https://docs.aws.amazon.com/managed-flink/latest/java/troubleshooting-async-deadlock-update-java-apps.html>
    flinkShadowJar(enforcedPlatform("software.amazon.awssdk:bom:2.20.144"))

    // Timestream sink connector
    flinkShadowJar files('libs/flink-connector-timestream-0.4-SNAPSHOT.jar')
    flinkShadowJar "org.projectlombok:lombok:1.18.22"
    flinkShadowJar "software.amazon.awssdk:aws-crt-client:${awsSDKVersion}"
    flinkShadowJar "software.amazon.awssdk:timestreamwrite:${awsSDKVersion}"
}

// make compileOnly dependencies available for tests:
sourceSets {
    main.compileClasspath += configurations.flinkShadowJar
    main.runtimeClasspath += configurations.flinkShadowJar

    test.compileClasspath += configurations.flinkShadowJar
    test.runtimeClasspath += configurations.flinkShadowJar

    javadoc.classpath += configurations.flinkShadowJar
}

run.classpath = sourceSets.main.runtimeClasspath

shadowJar {
    configurations = [project.configurations.flinkShadowJar]
}

Looking at the dependencies, I see that you have

flink-connector-kinesis

with your dependencies. Are you using Kinesis sink in your application? If so, kinesis connector shade AWS SDK classes, which needs to be replaced in order to resolve the issue.

We're using Kinesis source in our application! We haven't had any issues with that. I did not realize it could interfere with Timestream sink but it indeed might 🤔 Good catch, I will try the thing you linked.

Hmm, looks like the missing class exception is coming because of this file. There 2 options of addressing it: • exclude it as well, should be possible by not limiting exclusion by

*.class

for sdk classes. As far as I know this interceptor does not strictly necessary for correct functioning of sdk.

exclude("org/apache/flink/kinesis/shaded/software/amazon/awssdk/**/*.class")

->

exclude("org/apache/flink/kinesis/shaded/software/amazon/awssdk/**/*")

• exclude and replace with new version to point to new version of interceptor class.

Oh noes 😞 I already thought this would have solved the issue!