Hi Everybody. I have a Flink job deployed to Kubernetes trying to produce data to a AWS MSK Kafka topic, however, when I write to that topic, it never receives data. I tested my code using Docker Compose and Docker and it can send messages in that environment, but in the cloud deployment in Kubernetes, I notice the following. The cloud deployment in the metrics section shows the following metrics:

0.my_sink_name_my_group_id__Committer.numRecordsIn = 343
0.my_sink_name_my_group_id__Writer.numRecordsIn = 0

The writer metric is always zero. Consequently, that topic never receives any messages. In my functional Docker Compose deployment where Kafka messages are received the

Writer

metric is always

positive non-zero

. Does anybody understand what

Committer

and

Writer

do, and what I should consider to resolve this issue? This job uses three kafka topics. It uses AT_LEAST_ONCE delivery guarantee for Kafka, and EXACTLY_ONCE checkpointing. 1. kafka source 2. kafka sink for telegraf 3. kafka sink to statefun (but writing to that topic fails as described above). This particular Kafka cluster is running with only two bootstrap servers (in case that makes a difference).

Have you verified the connectivity from your flink job to the MSK? Are you using plaintext/TLS/IAM for data transport? In my experience, I've seen transport layer issues with MSK/Kafka where the errors get swallowed until an OOM error is thrown eventually due to a notorious kafka issue https://issues.apache.org/jira/browse/KAFKA-4090

HI Sharon. Thank you for the reply. I'll look into that. However, in this specific Flink job there are three Kafka Topics involved in this pipeline: 1. the source topic 2. a sink topic that flows to telegraf 3. another sink topic that flows to a Apache Flink Stateful Functions cluster. Only the third one fails, although that topic works externally to the job. I am concerned that this has something to do with Kafka transactions upon two different sinks. Maybe I need to assign transaction id prefixes when building the KafkaSink and also use exactly once semantics. I don't know though, I am just speculating.

I see. then its less likely to be an auth issue. Does the 3rd topic work with a different flink job? FWIW, the committer creates the checkpoint and the writer just commit /flush the records so there might be issues with creating checkpoints for the 3rd topic. You can also enable flink/kafka debug logging to gather more info. Curious to know what you’ve figured out.

Hi again. Thank you for helping me. I still have not figured out the solution, but I discovered that it works correctly when there is only one task manager, but fails to send when there are more than one task manager. This operator in question performs a reduction of the data with an event time tumbling window of "one millisecond." It's duty is to de-duplicate data that was previously processed in parallel with a multitude of keys. I do find it odd to have an event time window of 1 millisecond. Regardless, this operator does not receive data when there is more than one task manager. This "reduce" operator is the last one in the chain that flows to the kafka sink. When there is only one task manager, everything works, and all the sinks receive data. Although this is progress, I would like to scale my job and support multiple task managers.

Just a guess. It could be that you have a key skew issue (all the records are sent to 1 TM for the reduction operator). In this case, I think you can manually program the parallelism to be 1 for the reduction operation assume that you are using the data stream api.

This operator is a ProcessWindowFunction (maybe I should've used an aggregate reducer in-front of it, but I didn't), is keyed by an identifier. Wouldn't that hash allow for parallelism greater than one? Here is a simplified example of my code:

final DataStream<StateFunRequest> stateFunStream = dataStream.flatMap(new MapTags())
	.uid(Operators.MAP_TAGS.uid)
	.name(Operators.MAP_TAGS.operatorName)
	.keyBy(KeySelectors.keyByOrgAndUnitAndTagName())
	.process(new InterceptIdentifiers())
	.uid(Operators.INTERCEPT_INDENTIFERS.uid)
	.name(Operators.INTERCEPT_INDENTIFERS.operatorName)
	.keyBy(KeySelectors.keyByOrgAndUnitAndId())
	.window(TumblingEventTimeWindows.of(Time.milliseconds(1)))
	.process(new ReduceByIdentifiers())
	.uid(Operators.REDUCE_BY_IDENTIFERS.uid)
	.name(Operators.REDUCE_BY_IDENTIFERS.operatorName);
			
	stateFunStream.sinkTo(stateFunIngress);

Have you checked the metrics for each operator to see where it stopped to get data? My suggestion was to move

.setParallelism(1)

after

keyBy

as my guess was the all keys get sent to the same TM because of the key skew. Window operator needs watermark and if there is no data, the watermark can't progress.

Yes, I checked the metrics. They are quite suspicious:

0.my_sink_name_my_group_id__Committer.numRecordsIn = 343
0.my_sink_name_my_group_id__Writer.numRecordsIn = 0

It’s committing but not writing. However, I think your premise is wrong that all task managers need to receive data. I could be wrong, but my understanding is that if data keeps on arriving and there is data-skew then that introduces hot-spot related performance issues, but the ascending event time watermark will act like a string pulling time forward across all task managers. In fact, I have observed this behavior in other Flink jobs that have data-skew, and everything worked fine. What do you think?

Cool. Glad that you’ve figured it out 🙌