Hi everyone, I'm running Apache Flink with the Kubernetes operator. The image I'm using in my deployment contains the following jar "https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/1.17.1/flink-sql-connector-kafka-1.17.1.jar". In my job I'm using the connector "upsert-kafka".

CREATE TABLE `MyTable` (
  `object_ID` INTEGER PRIMARY KEY NOT ENFORCED,
  `Path` VARCHAR,
  `Status` VARCHAR NOT NULL
) WITH (
  'connector' = 'upsert-kafka',
  'value.format' = 'json',
  'topic' = 'my-topic',
  'key.format' = 'json',
  'properties.security.protocol' = 'SASL_SSL',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.bootstrap.servers' = 'xxxx',
  'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="xxxx" password="yyyy";'
);

My pod complains that it can't find the "PlainLoginModule". I found this documentation: https://issues.apache.org/jira/browse/FLINK-31361 But using

<http://org.apache.flink.kafka.shaded.org|org.apache.flink.kafka.shaded.org>.apache.kafka.common.security.plain.PlainLoginModule

instead of

org.apache.kafka.common.security.plain.PlainLoginModule

doesn't solve the issue. What is the correct name? Am I missing something else? (I've tried the workaround suggested in the link above; but that leads to other errors then 🙃)

It should

<http://org.apache.flink.kafka.shaded.org|org.apache.flink.kafka.shaded.org>.apache.kafka.common.security.plain.PlainLoginModule

Thanks @Martijn Visser! I actually also want to connect to Confluent Cloud

I use

flink-sql-connector-kafka-3.0.0-1.17.jar

which is in my

lib

folder.

👍 thanks for sharing your setup @Martijn Visser ! While writing everything down i'm wondering now if I'm using the correct jar actually 🤔 In my first attempt with Flink (my goal is to achieve MSSQL -> Flink -> Kafka) i was using Debezium+Kafka between MSSQL and Flink. Hence i was using the lib

flink-sql-connector-kafka

. But now i'm actually trying to use

flink-sql-connector-sqlserver-cdc

to skip on my own the Debezium+Kafka part. So maybe the lib i need to talk to kafka is not anymore

flink-sql-connector-kafka

but simply the lib

flink-connector-kafka

The SQL connector Kafka is a shaded version of the connector which includes the Kafka Client. If you use the regular one, you will also need to add the necessary Kafka dependencies

🤔 Now i'm more confused 😄 Which library-ies do I need to use the "upsert-kafka" connector?

I believe because Flink 1.17 uses a different version of Kafka Clients, let me check...

@Martijn Visser I changed the kafka version and got rid of the previous error, thanks! 🎉 Now i'm using:

kafka-clients/3.2.3
flink-connector-kafka/3.0.0-1.17

If I may ask another hint from you....what do you think of this error message; any ideas/pointers for me to solve it? 👀

│ Caused by: java.lang.ClassCastException: cannot assign instance of org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.OffsetResetStrategy to field org.apache.flink.connector.kafka.source.enumerator.initializer.ReaderHandledOffsetsInitializer.offsetResetStrategy of type org.apache.kafka.clients.consumer.OffsetResetStrategy in instance of org.apache.flink.connector.kafka.source.enumerator.initializer.ReaderHandledOffsetsInitializer

In my use case actually I don't consume anything from Kafka, just produce to kafka; so this error surprises me 🤷

Is there a particular reason why you aren't using

flink-sql-connector-kafka/3.0.0-1.17

?

@Tiphanie Dousset facing similar issue. were you able to resolve this issue ?

Hi @Martijn Visser No specific reason why

flink-connector-kafka

instead of

flink-sql-connector-kafka/3.0.0-1.17

. I thought the first one would be enough since i need to use the kafka connector to "just" stream my Flink Table into Kafka (independently of sql). 🤷 But anyway, after trying some different combination; the one working is the following. FYI @Rajat Ahuja : • In my Docker file that will be use as image in my Kubernetes deployment:

RUN wget -P /opt/flink/lib <https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/1.17.1/flink-sql-connector-kafka-1.17.1.jar> \
    <https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-sqlserver-cdc/2.4.0/flink-sql-connector-sqlserver-cdc-2.4.0.jar> \
    <https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.2.3/kafka-clients-3.2.3.jar> \
    <https://repo1.maven.org/maven2/org/apache/flink/flink-connector-base/1.17.1/flink-connector-base-1.17.1.jar> \
    && wget -P /opt/flink/plugins <https://repo1.maven.org/maven2/org/apache/flink/flink-s3-fs-hadoop/1.17.1/flink-s3-fs-hadoop-1.17.1.jar>

• My K8 deployment:

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:  
  name: xxxx
  namespace: yyyy
spec:
  flinkConfiguration:
    containerized.master.env.ENABLE_BUILT_IN_PLUGINS: flink-s3-fs-hadoop-1.17.1.jar
    containerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS: flink-s3-fs-hadoop-1.17.1.jar
    execution.checkpointing.interval: 10s
    fs.s3a.aws.credentials.provider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
    high-availability.storageDir: <s3://bbbb/>
    high-availability.type: kubernetes
    jobmanager.memory.flink.size: 460m
    jobmanager.memory.jvm-overhead.min: 3m
    kubernetes.container.image.pull-secrets: aaaaa
    state.backend.incremental: "true"
    state.backend.type: rocksdb
    state.checkpoints.dir: <s3://zzzzzz/>
    state.savepoints.dir: <s3://yyyyy/>
    taskmanager.memory.flink.size: 460m
    taskmanager.memory.jvm-overhead.min: 3m
    taskmanager.memory.managed.size: 120m
    taskmanager.numberOfTaskSlots: "2"
  flinkVersion: v1_17
  image: <path-to-the-registry-with-image-built-from-Dockerfile-bove>
  imagePullPolicy: Always
  job:
    args:
    - /opt/sql-scripts/simple.sql
    jarURI: local:///opt/flink/usrlib/sql-runner.jar
    parallelism: 1
    savepointTriggerNonce: 0
    state: running
    upgradeMode: stateless
  jobManager:
    podTemplate:
      apiVersion: v1
      kind: Pod
      spec:
        containers:
        - name: flink-main-container
          volumeMounts:
          - mountPath: /opt/sql-scripts
            name: sql-scripts
        imagePullSecrets:
        - name: eeeeeee
        volumes:
        - configMap:
            name: sql-scripts
          name: sql-scripts
    replicas: 1
    resource:
      cpu: 0.25
      memory: 760m
  serviceAccount: flink
  taskManager:
    resource:
      cpu: 0.25
      memory: 760m

NB: right now i have

upgradeMode: stateless

but i plan on moving to

upgradeMode: last-state

I thought the first one would be enough since i need to use the kafka connector to "just" stream my Flink Table into Kafka (independently of sql).

@Tiphanie Dousset If you're using SQL, I would recommend using the SQL version of the Kafka connector. That's because it already contains the correctly shaded JARs to connect to Kafka (the Client etc). Then you're always certain that you have a bundle that has been working from a Flink POV during testing