[FLINK-4616] Added functionality through which watermarks for each partition are saved and loaded via checkpointing mechanism #3031

MayerRoman · 2016-12-20T07:48:27Z

[FLINK-4616] Kafka consumer doesn't store last emmited watermarks per partition in state.

…rtition are saved and loaded via checkpointing mechanism

MayerRoman · 2016-12-22T09:01:52Z

I think that the changes that I propose eliminates the possibility of starting with checkpoints created before my code changes.

Because now it saves ListState<Tuple2<KafkaTopicPartition, Tuple2<Long, Long>>> (partition + offset + watermark).
And before it saved ListState<Tuple2<KafkaTopicPartition, Long>> (partition + offset).

(I mean checkpoints version later then 1.1.
Recently added backward compatibility with 1.1 snapshots is taken into account in my commit, with it, I think, everything is ok)

Please advise me how to repair backward compatibility.

I have some idea of how to implement it:

somehow verify returned from stateStore.getSerializableListState(..) object
in initializeState method
https://github.com/apache/flink/pull/3031/files?diff=unified#diff-06bf4a7f73d98ef91309154654563475R321

is it
ListState<Tuple2<KafkaTopicPartition, Long>>
or
ListState<Tuple2<KafkaTopicPartition, Tuple<Long, Long>>>

Use for saving watermark separate state-object.

Or it is necessary implement different way.

I would be grateful for help.

tzulitai · 2017-01-23T08:26:02Z

Thank you for the contribution @MayerRoman. Just want to let you know that I've noticed this PR, and I think the issue is definitely something we want to fix. I'll allocate some time this week to review the PR.

MayerRoman · 2017-01-25T07:15:05Z

Thank you Tzu-Li Tai.

tzulitai

@MayerRoman I've done a first pass review, I'll definitely do a second detailed pass once we sort the following problems out:

I think it'll make sense to have a new specific state class that wraps a topic-partition's offset and watermark as checkpointed state.
This feature is expected to be fixed for 1.3.0, which means savepoints taken during 1.2 and 1.1 will be considered as old version checkpoints. Therefore, I think the original Tuple2<KafkaTopicPartition, Long> state will first be passed through the old restoreState() method. Then, in initializeState(), we can try to bridge the state type change from the old one to the new one. I'm still confirming this though, as this method call pattern for old version state might only be the case for 1.1 -> 1.2, and not for 1.2 -> 1.3.
There's an important thing missing in this change: after a restore with watermarks, when starting to run the source, you'll need to check if a watermark should first be emitted from the source considering the already existing watermarks across its subscribed partitions. I don't think this PR has addressed this yet.

Let me know what you think about the above :-)

tzulitai · 2017-01-25T11:17:22Z

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

 	private SerializedValue<AssignerWithPunctuatedWatermarks<T>> punctuatedWatermarkAssigner;

-	private transient ListState<Tuple2<KafkaTopicPartition, Long>> offsetsStateForCheckpoint;
+	private transient ListState<Tuple2<KafkaTopicPartition, Tuple2<Long, Long>>> offsetsAndWatermarksStateForCheckpoint;


I think we should switch to have a specific checkpointed state object instead of continuing to "extend" the original Tuple. This will also be helpful for compatibility for any future changes to the checkpointed state.

tzulitai · 2017-01-25T11:21:41Z

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

-								Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
+					for (Map.Entry<KafkaTopicPartition, Tuple2<Long, Long>> kafkaTopicPartitionOffsetAndWatermark : restoreToOffsetAndWatermark.entrySet()) {
+						offsetsAndWatermarksStateForCheckpoint.add(
+								Tuple2.of(kafkaTopicPartitionOffsetAndWatermark.getKey(), kafkaTopicPartitionOffsetAndWatermark.getValue()));


Having a specific checkpoint state object will also be helpful for code readability in situations like this one (it's quite tricky to understand quickly what the key / value refers to, as well as some of the f0, f1 calls in other parts of the PR. I know the previous code used f0 and f1 also, but I think it's a good opportunity to improve that).

tzulitai · 2017-01-26T06:06:34Z

...ase/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java

+
+			case NO_TIMESTAMPS_WATERMARKS: {
+
+				for (KafkaTopicPartitionState<KPH> partition : allPartitions) {


Excessive empty line above this line.

tzulitai · 2017-01-26T06:07:30Z

...ase/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java

+
+			default:
+				// cannot happen, add this as a guard for the future
+				throw new RuntimeException();


Would be good to have a reason message here.

tzulitai · 2017-01-26T06:07:52Z

...ase/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java

+
+			default:
+				// cannot happen, add this as a guard for the future
+				throw new RuntimeException();


Would be good to have a reason message here.

tzulitai · 2017-01-26T06:11:58Z

...ink/streaming/connectors/kafka/internals/KafkaTopicPartitionStateWithPeriodicWatermarks.java

 		return partitionWatermark;
 	}

+	void setCurrentWatermarkTimestamp(long watermarkTimestamp) {


The other methods seem to be public (although they can actually be package-private). Should we stay consistent with that here?

tzulitai · 2017-01-26T08:36:03Z

A re-clarification about backwards compatibility for state type change:
Currently, one limitation for compatible applications across savepoint restore is that you can't change the type of state otherwise state restore will fail, therefore not compatible. The only work around, is to have another field as the new state with the new type, and somehow try to "encode" / "decode" the watermark state into / from the original Tuple2<KafkaTopicPartition, Long>. I don't think this is easily possible ...

At the same time, there was recent discussion about letting the window operators also checkpoint watermarks. So perhaps we might not need to let the Kafka sources checkpoint watermarks in the end, if the window operators already take care of restoring the previous event time.
What I'm curious about right now is whether or not in the future, redistributions of Kafka partition states across source subtasks will work well with the checkpointed watermarks in the downstream window operators. I don't think there should be a problem, i.e. late elements should still be dropped by the window operators in that case.

@aljoscha can you perhaps help clarify this?

rmetzger · 2017-02-12T15:42:19Z

I think this PR should also include a test for the added feature.

MayerRoman · 2017-02-13T07:52:37Z

I hope that I ended up with another issue, and I come back to this.

First, I want to ask a question that perhaps remove all the others.

Tzu-Li Tai, did I understand correctly that if discussion about letting the window operators checkpoint watermarks lead to the decision to implement this functionality in the window operators, the need to preserve the state of watermarks in Kafka consumer will disappear?

tzulitai · 2017-02-13T08:30:11Z

Hi @MayerRoman! Thank you for coming back to this issue.

I had a quick chat offline with @aljoscha about whether or not it'll be reasonable to add this. Either your approach in this PR or letting window operators checkpoint watermarks will both solve the issue of late elements after restore. However, we thought that we should let the user function code be free of responsibility of checkpointing watermarks, and let user code simply leave that to the streaming internals (checkpointed by window operators, and perhaps also by source operators).

So, essentially, the Kafka consumer should not need to checkpoint watermarks, and we can close this PR and the JIRA ticket. Very sorry for the late discussion on this, and having you worked on it already.

Let we know what you think and whether or not you agree :-)

MayerRoman · 2017-02-13T12:47:36Z

Hello, Tzu-Li Tai!
I think you make a good decision and agree to the fact that the PR and JIRA ticket can be closed.

Do not worry about the done work, I got good experience in the process.

tzulitai · 2017-02-13T14:57:54Z

Thanks. Can you please close this PR then :-D ? I'll close the JIRA.

MayerRoman · 2017-02-13T16:07:14Z

Ok

MayerRoman changed the title ~~[FLINK-4616] Added functionality through which watermarks for each pa…~~ [FLINK-4616] Added functionality through which watermarks for each partition are saved and loaded via checkpointing mechanism Dec 20, 2016

[FLINK-4616] Added functionality through which watermarks for each pa…

b79b580

…rtition are saved and loaded via checkpointing mechanism

MayerRoman force-pushed the FLINK_4616 branch from bb9a549 to b79b580 Compare December 21, 2016 17:04

tzulitai requested changes Jan 26, 2017

View reviewed changes

MayerRoman closed this Feb 13, 2017

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


		case NO_TIMESTAMPS_WATERMARKS: {

		for (KafkaTopicPartitionState<KPH> partition : allPartitions) {

[FLINK-4616] Added functionality through which watermarks for each partition are saved and loaded via checkpointing mechanism #3031

[FLINK-4616] Added functionality through which watermarks for each partition are saved and loaded via checkpointing mechanism #3031

Uh oh!

Conversation

MayerRoman commented Dec 20, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

MayerRoman commented Dec 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

tzulitai commented Jan 23, 2017

Uh oh!

MayerRoman commented Jan 25, 2017

Uh oh!

tzulitai left a comment

Choose a reason for hiding this comment

Uh oh!

tzulitai Jan 25, 2017

Choose a reason for hiding this comment

Uh oh!

tzulitai Jan 25, 2017

Choose a reason for hiding this comment

Uh oh!

tzulitai Jan 26, 2017

Choose a reason for hiding this comment

Uh oh!

tzulitai Jan 26, 2017

Choose a reason for hiding this comment

Uh oh!

tzulitai Jan 26, 2017

Choose a reason for hiding this comment

Uh oh!

tzulitai Jan 26, 2017

Choose a reason for hiding this comment

Uh oh!

tzulitai commented Jan 26, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rmetzger commented Feb 12, 2017

Uh oh!

MayerRoman commented Feb 13, 2017

Uh oh!

tzulitai commented Feb 13, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

MayerRoman commented Feb 13, 2017

Uh oh!

tzulitai commented Feb 13, 2017

Uh oh!

MayerRoman commented Feb 13, 2017

Uh oh!

Uh oh!

MayerRoman commented Dec 20, 2016 •

edited

Loading

MayerRoman commented Dec 22, 2016 •

edited

Loading

tzulitai commented Jan 26, 2017 •

edited

Loading

tzulitai commented Feb 13, 2017 •

edited

Loading