Unrecognized SASL ClientCallback using PyFlink and Kafka

[aguaguilar1907]

Hello, I'm trying to use AWS managed Flink with pyflink. The purpose of this is to consume data from a kinesis, do some transformation and then load the data into redshift.

So, I'm using the following pom.xml in order to connect to MSK using IAM role.

<project>
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-iam-sql-kafka-connector</artifactId>
    <version>1</version>

    <properties>
        <flink.version>1.18.0</flink.version>
    </properties>

    <!-- https://docs.aws.amazon.com/managed-flink/latest/java/troubleshooting-async-deadlock-update-java-apps.html --> 
    <dependencyManagement>
        <dependencies>

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

        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>3.1.0-1.18</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/software.amazon.msk/aws-msk-iam-auth -->
        <dependency>
            <groupId>software.amazon.msk</groupId>
            <artifactId>aws-msk-iam-auth</artifactId>
            <version>2.1.0</version>
        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.4.1</version>
                <configuration>
                    <transformers>
                        <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                    </transformers>
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

        </plugins>

    </build>

</project>

The problem that I'm seeing is: there is no chance to connect to MSK using the following properties:

properties = {
        "bootstrap.servers": "myserver:9098",
        "group.id": "packing_mib_stats",
        "security.protocol": "SASL_SSL",
        "sasl.mechanism": "AWS_MSK_IAM",
        "sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
        "sasl.login.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
    }

    stream = env.add_source(FlinkKafkaConsumer("core_events.execution", SimpleStringSchema(), properties))
    stream.key_by(lambda x: x) \
        .process(ProcessFunction())

And the error is:

org.apache.kafka.common.errors.SaslAuthenticationException: An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: Exception while evaluating challenge [Caused by javax.security.auth.callback.UnsupportedCallbackException: Unrecognized SASL ClientCallback]) occurred when evaluating SASL token received from the Kafka Broker. Kafka Client will go to AUTHENTICATION_FAILED state.
Caused by: javax.security.sasl.SaslException: Exception while evaluating challenge [Caused by javax.security.auth.callback.UnsupportedCallbackException: Unrecognized SASL ClientCallback]

Any idea what could happening? I'm using Flink 1.18.0. Thanks!

[aguaguilar1907]

Closed given it's duplicated by https://github.com/aws/aws-msk-iam-auth/issues/11

[github-actions[bot]]

This issue is now closed. Comments on closed issues are hard for our team to see. If you need more assistance, please open a new issue that references this one.