|
|
Dear Flink Community,
When using exactly once checkpointing along kafka and connecting Kafka using SASL_SSL the job gets stuck with status of Kafka producer in Initialising state for around 10 mins and post that it functions normally across multiple checkpoints. Following are various config and observations.
Flink/Job config:
VERSION: Flink 1.13.1 for both env and jars.
1. Single node with checkpointing in exactly once mode and checkpointing interval as 2 minutes. ( tried with 10 minutes still same issue)
2. The job has source as flink kafka consumer for topic 1 and sink as flink kafka producer for topic 2 and parallelism as 4 for source & sink.
3. The kafka producer uses FlinkKafkaProducer implementation in EXACTLY_ONCE mode and connects to broker using SASL_SSL where SSL certificate is self signed.
Kafka config:
VERSION: 2.8 for broker. For Flink Job tried 2.4.1 (shipped with flink kafka connector) and 2.8.0
1. Single broker with both topics having 4 partitions.
2. Everything works correctly using Kafka Consumer commands when using SASL_SSL mode.
Observation for debug logs:
Background: In dev env with PLAIN_TEXT connection everything was working perfectly. When I changed PLAIN_TEXT to SASL_SSL I was get timeout exceptions. Which was resolved after I increased max.block.ms from 60000 to 90000 but then started facing delayed start issue.
1. About error/exception:
I did not find any exception in debug logs but the notified following pattern which keeps repeating until successful initialisation
SSL_HANDSHAKE --> SCRAM_AUTHENTICATION —> FIND_COORDINATOR —> FETCH PRODUCER_ID —> Transition from state INITIALIZING to READY —> CLOSE PRODUCER
The producer keep repeating above step until following pattern occurs
SSL_HANDSHAKE --> SCRAM_AUTHENTICATION —> FIND_COORDINATOR —> FETCH PRODUCER_ID —> Transition from state INITIALIZING to READY —> READY to IN_TRANSACTION.
Also while kafka producer/sink is stuck in Initialising state if I cancel job the kafka sink operator gets stuck and I get following error Task did not exit gracefully within 180 + seconds. And task manager crashes.
2. Turning off SSL and using SASL_PLAINTEXT - When connecting to kafka broker using SASL_PLAINTEXT mode the kafka producer gets initialised and starts processing data in 50-60 secs.
3. Disable checkpointing - When checkpointing is disabled and use SASL_SSL/SASL_PLAINTEXT mode fro kafka. The kafka producer get initialised immediately but Kafka Consumer or sink takes 10-15 secs to initialise and then app operators start processing data.
4. If wrote a producer directly using kafka producer from kafka library. And added transaction to it. My obeservation was kafkaProducer.initTransactions() method takes around 30 sec to execute post which it starting publishing data to kafka topic.
5. Same pattern of delayed start seems to appear incase of crash and automatic restart of job.
6. No exception in Flink Dashboard UI for job.
PFA, a copy of part of log.
|
Locked
