[FLINK-9303] [kafka] Adding support for unassign dynamically partitions from kafka consumer when they become unavailable #5991

EAlexRojas · 2018-05-11T13:36:33Z

What is the purpose of the change

This pull request add an option on the kafka consumer to check for unavailable partitions and unassign them from the consumer. That way the consumer does not request for records on invalid partitions and prevent Logs noises.

Brief change log

Modify the partition discovery system to check not only new partitions, but also check partitions that are no longer available.
Check for partitions no longer available recovered from state.
Add option on kafka consumer to activate this checks

Verifying this change

This change added tests and can be verified as follows:
Manually verified as follows:

Create a job with a kafka consumer listening to a topic pattern and having partition discovery activated and the property introduced in this PR set to true.
Configure Kafka to have set the following properties:
delete.topic.enable=true
auto.create.topics.enable=false
Create some topics matching the pattern.
Run the job.
While running, remove some of the topics.
Verify the partitions are unassigned and the job continue running without Log noises.

I guess this can be tested with e2e tests, but I'm not familiarised with the system in place

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)

tedyu · 2018-05-11T15:21:10Z

...afka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java

@@ -221,7 +221,8 @@ private FlinkKafkaConsumer08(
 				getLong(
 					checkNotNull(props, "props"),
 					KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, PARTITION_DISCOVERY_DISABLED),
-				!getBoolean(props, KEY_DISABLE_METRICS, false));
+				!getBoolean(props, KEY_DISABLE_METRICS, false),
+				getBoolean(props, KEY_CHECK_UNAVAILABLE_TOPICS, false));


Should this be named KEY_CHECK_UNAVAILABLE_PARTITIONS ?

You're right, I'll change it

tedyu · 2018-05-11T15:23:13Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

@@ -80,6 +82,9 @@
 	/** The queue of unassigned partitions that we need to assign to the Kafka consumer. */
 	private final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue;

+	/** The list of partitions to be removed from kafka consumer. */
+	private final List<TopicPartition> partitionsToBeRemoved;


Should this be Set to facilitate fast lookup ?

You are right, a Set should be better for all the calls to the contains() method.

tedyu · 2018-05-11T15:25:24Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

-	void reassignPartitions(List<KafkaTopicPartitionState<TopicPartition>> newPartitions) throws Exception {
-		if (newPartitions.size() == 0) {
+	void reassignPartitions(List<KafkaTopicPartitionState<TopicPartition>> newPartitions, List<TopicPartition> partitionsToBeRemoved) throws Exception {
+		if (newPartitions.size() == 0 && partitionsToBeRemoved.isEmpty()) {


size() == 0 -> isEmpty

…ns from kafka consumer when they become unavailable - Check for unavailable partitions recovered from state - Using kafka consumer option to activate this validations

EAlexRojas · 2018-05-22T12:11:28Z

PR updated taking into account comments

tzulitai

Thanks for the PR @EAlexRojas.
I had a shallow review, with some questions on higher-level design aspects of the change. Please let me know what you think.

tzulitai · 2018-05-23T07:44:21Z

...a-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java

@@ -235,7 +243,8 @@ public FlinkKafkaConsumerBase(
 			Pattern topicPattern,
 			KeyedDeserializationSchema<T> deserializer,
 			long discoveryIntervalMillis,
-			boolean useMetrics) {
+			boolean useMetrics,
+			boolean checkUnavailablePartitions) {


Why do we want this to be configurable? Is there any case that we would prefer to leave them untouched?

I did it in that way only because this is something new, so I though that maybe you may want it to be configurable. But you are right I cannot think of a case we would prefer to keep the unavailable partitions.
I'll update the PR to make it the default behaviour if it's ok for you.

tzulitai · 2018-05-23T07:52:32Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

@@ -374,8 +385,8 @@ void setOffsetsToCommit(
 	 * <p>This method is exposed for testing purposes.
 	 */
 	@VisibleForTesting
-	void reassignPartitions(List<KafkaTopicPartitionState<TopicPartition>> newPartitions) throws Exception {
-		if (newPartitions.size() == 0) {
+	void reassignPartitions(List<KafkaTopicPartitionState<TopicPartition>> newPartitions, Set<TopicPartition> partitionsToBeRemoved) throws Exception {


I have the feeling that this method is way too complex now, to a point that it might make more sense to break this up into 2 different methods - addPartitionsToAssignment and removePartitionsFromAssignment.

I though about it, but my only concern is the case where we'd have both, partitions to add and partitions to remove...
the consumerCallBridge.assignPartitions() takes the whole new list of partitions, so in that case, we would need to wait for the first assignment (e.g. add new partitions) before doing the second assignment (e.g. remove partitions) in order to have a consistent list of partitions.
I think we would try to have only one call to consumerCallBridge.assignPartitions().

Maybe I could refactor the part where partitions are removed from old partitions to a separate private method like removeFromOldPartitions() ?

What do you think ?

tzulitai · 2018-05-23T07:56:22Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

@@ -80,6 +83,9 @@
 	/** The queue of unassigned partitions that we need to assign to the Kafka consumer. */
 	private final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue;

+	/** The list of partitions to be removed from kafka consumer. */
+	private final Set<TopicPartition> partitionsToBeRemoved;


Would it actually make more sense that we have a queue for this? Like how we are handling unassigned new partitions via the unassignedPartitionsQueue. The fact that this is a set means that we will need to eventually remove entries from it anyways.

From my understanding, for unassigned partitions we can use a Queue because it does not matter which consumer will take the new partitions.
But we can not use a Queue for partitions to be removed because we only can remove the partitions from the consumer that is actually subscribed to that partition.
Does that make sense ?

EAlexRojas · 2018-05-24T10:00:19Z

.../src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java

@@ -240,7 +249,9 @@ public void run() {
 						newPartitions = unassignedPartitionsQueue.getBatchBlocking();
 					}
 					if (newPartitions != null) {
-						reassignPartitions(newPartitions);
+						reassignPartitions(newPartitions, new HashSet<>());


I just realized this should be actually
reassignPartitions(newPartitions, partitionsToBeRemoved);

aljoscha · 2020-07-09T12:21:52Z

I'm closing this as "Abandoned", since there is no more activity and the code base has moved on quite a bit. Please re-open this if you feel otherwise and work should continue.

tedyu reviewed May 11, 2018

View reviewed changes

[FLINK-9303] [kafka] Adding support for unassign dynamically partitio…

a7f23f9

…ns from kafka consumer when they become unavailable - Check for unavailable partitions recovered from state - Using kafka consumer option to activate this validations

EAlexRojas force-pushed the kafka-unassign-partitions-fix branch from a17d0dc to a7f23f9 Compare May 22, 2018 08:15

tzulitai reviewed May 23, 2018

View reviewed changes

EAlexRojas commented May 24, 2018

View reviewed changes

rmetzger added the component=Connectors/Kafka label Mar 18, 2019

aljoscha closed this Jul 9, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-9303] [kafka] Adding support for unassign dynamically partitions from kafka consumer when they become unavailable #5991

[FLINK-9303] [kafka] Adding support for unassign dynamically partitions from kafka consumer when they become unavailable #5991

EAlexRojas commented May 11, 2018

tedyu May 11, 2018

EAlexRojas May 16, 2018

tedyu May 11, 2018

EAlexRojas May 16, 2018

tedyu May 11, 2018 •

edited

EAlexRojas commented May 22, 2018

tzulitai left a comment

tzulitai May 23, 2018

EAlexRojas May 24, 2018

tzulitai May 23, 2018

EAlexRojas May 24, 2018

tzulitai May 23, 2018

EAlexRojas May 24, 2018

EAlexRojas May 24, 2018

aljoscha commented Jul 9, 2020

[FLINK-9303] [kafka] Adding support for unassign dynamically partitions from kafka consumer when they become unavailable #5991

[FLINK-9303] [kafka] Adding support for unassign dynamically partitions from kafka consumer when they become unavailable #5991

Conversation

EAlexRojas commented May 11, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tedyu May 11, 2018 • edited

Choose a reason for hiding this comment

EAlexRojas commented May 22, 2018

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aljoscha commented Jul 9, 2020

tedyu May 11, 2018 •

edited