How does Flink manage the kafka offset

[Eleanore Jin]

Hi there, 

We are running apache beam application with flink being the runner. 

We use the KafkaIO connector to read from topics: https://beam.apache.org/releases/javadoc/2.19.0/

and we have the following configuration, which enables auto commit of offset, no checkpointing is enabled, and it is performing element wise processing.

So we run our application in Flink Job Cluster mode, and if I run the same job twice, meaning start 2 flink job clusters, then I see message being processed twice. 

My understanding is, upon startup, Flink Job Manager will contact kafka to get the offset for each partition for this consume group, and distribute the task to task managers, and it does not use kafka to manage the consumer group. 

and when the 2nd job cluster starts up, it does the same thing, so the 1st job cluster is not aware of there are new consumers from the same consume group have joined. 

But if I add more task managers to the same job cluster, then job manager is aware of more consumers from this consume group has joined, and it will rebalance the partition consumption if needed.

Is my understanding correct?

Thanks a lot!

Eleanore

Map<String, Object> consumerConfig = ImmutableMap.<String, Object>builder()
.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup)
.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true)
.put(ConsumerConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG")
.build();

return KafkaIO.<String, JsonNode>read()
.withBootstrapServers(kafkaSettings.getBootstrapServers())
.withTopic(topic)
.withKeyDeserializer(KeyDeserializer.class)
.withValueDeserializerAndCoder(getDeserializer(encoding), new JsonNodeCoder<>())
.withConsumerConfigUpdates(consumerConfig)
.withoutMetadata();

[Benchao Li]

Hi Jin,

See below inline replies:

My understanding is, upon startup, Flink Job Manager will contact kafka to get the offset for each partition for this consume group, and distribute the task to task managers, and it does not use kafka to manage the consumer group. 

Generally, yes. If you are not using checkpoint, and starting from group-offsets, Flink will read offset from Kafka at startup.

and when the 2nd job cluster starts up, it does the same thing, so the 1st job cluster is not aware of there are new consumers from the same consume group have joined. 

Yes.

But if I add more task managers to the same job cluster, then job manager is aware of more consumers from this consume group has joined, and it will rebalance the partition consumption if needed.

No. Flink does not rebalance the partitions when new task managers joined cluster. It only did so when job restarts and job parallelism changes.

Hope it helps.

Jin Yi <[hidden email]> 于2020年2月21日周五 上午6:14写道：

Hi there, 

We are running apache beam application with flink being the runner. 

We use the KafkaIO connector to read from topics: https://beam.apache.org/releases/javadoc/2.19.0/

and we have the following configuration, which enables auto commit of offset, no checkpointing is enabled, and it is performing element wise processing.

So we run our application in Flink Job Cluster mode, and if I run the same job twice, meaning start 2 flink job clusters, then I see message being processed twice. 

My understanding is, upon startup, Flink Job Manager will contact kafka to get the offset for each partition for this consume group, and distribute the task to task managers, and it does not use kafka to manage the consumer group. 

and when the 2nd job cluster starts up, it does the same thing, so the 1st job cluster is not aware of there are new consumers from the same consume group have joined. 

But if I add more task managers to the same job cluster, then job manager is aware of more consumers from this consume group has joined, and it will rebalance the partition consumption if needed.

Is my understanding correct?

Thanks a lot!

Eleanore

Map<String, Object> consumerConfig = ImmutableMap.<String, Object>builder()
.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup)
.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true)
.put(ConsumerConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG")
.build();

return KafkaIO.<String, JsonNode>read()
.withBootstrapServers(kafkaSettings.getBootstrapServers())
.withTopic(topic)
.withKeyDeserializer(KeyDeserializer.class)
.withValueDeserializerAndCoder(getDeserializer(encoding), new JsonNodeCoder<>())
.withConsumerConfigUpdates(consumerConfig)
.withoutMetadata();

--

[Eleanore Jin]

Hi Benchao,

Thanks a lot!

Eleanore

On Thu, Feb 20, 2020 at 4:30 PM Benchao Li <[hidden email]> wrote:

Hi Jin,

See below inline replies:

My understanding is, upon startup, Flink Job Manager will contact kafka to get the offset for each partition for this consume group, and distribute the task to task managers, and it does not use kafka to manage the consumer group. 

Generally, yes. If you are not using checkpoint, and starting from group-offsets, Flink will read offset from Kafka at startup.

and when the 2nd job cluster starts up, it does the same thing, so the 1st job cluster is not aware of there are new consumers from the same consume group have joined. 

Yes.

But if I add more task managers to the same job cluster, then job manager is aware of more consumers from this consume group has joined, and it will rebalance the partition consumption if needed.

No. Flink does not rebalance the partitions when new task managers joined cluster. It only did so when job restarts and job parallelism changes.

Hope it helps.

Jin Yi <[hidden email]> 于2020年2月21日周五 上午6:14写道：

Hi there, 

We are running apache beam application with flink being the runner. 

We use the KafkaIO connector to read from topics: https://beam.apache.org/releases/javadoc/2.19.0/

and we have the following configuration, which enables auto commit of offset, no checkpointing is enabled, and it is performing element wise processing.

So we run our application in Flink Job Cluster mode, and if I run the same job twice, meaning start 2 flink job clusters, then I see message being processed twice. 

My understanding is, upon startup, Flink Job Manager will contact kafka to get the offset for each partition for this consume group, and distribute the task to task managers, and it does not use kafka to manage the consumer group. 

and when the 2nd job cluster starts up, it does the same thing, so the 1st job cluster is not aware of there are new consumers from the same consume group have joined. 

But if I add more task managers to the same job cluster, then job manager is aware of more consumers from this consume group has joined, and it will rebalance the partition consumption if needed.

Is my understanding correct?

Thanks a lot!

Eleanore

Map<String, Object> consumerConfig = ImmutableMap.<String, Object>builder()
.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup)
.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true)
.put(ConsumerConfig.METRICS_RECORDING_LEVEL_CONFIG, "DEBUG")
.build();

return KafkaIO.<String, JsonNode>read()
.withBootstrapServers(kafkaSettings.getBootstrapServers())
.withTopic(topic)
.withKeyDeserializer(KeyDeserializer.class)
.withValueDeserializerAndCoder(getDeserializer(encoding), new JsonNodeCoder<>())
.withConsumerConfigUpdates(consumerConfig)
.withoutMetadata();

--

Benchao Li
School of Electronics Engineering and Computer Science, Peking University
Tel:[phone]
Email: [hidden email]; [hidden email]