GitHub user kl0u opened a pull request:

https://github.com/apache/flink/pull/3945

FLINK-6634 [cep] NFASerializer serializes ComputationState counter.

Before the recently introduces `NFASerializer` was not serializing the `counter` of the `ComputationState`. This could result in exceptions thrown when trying to put a new element in the `SharedBuffer` after restoring from a failure or when using RocksDB.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/kl0u/flink cep-nfa-serializer

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/3945.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3945

commit 5878526fc75d484705bdda95d97e4b711e9dbc17
Author: kkloudas <kkloudas@gmail.com>
Date: 2017-05-19T08:40:44Z

FLINK-6634 [cep] NFA serializer serializes ValueTimerWrapper counter.

Github user zentol commented on a diff in the pull request:

https://github.com/apache/flink/pull/3945#discussion_r117476822

— Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java —
@@ -930,6 +930,7 @@ public void serialize(NFA<T> record, DataOutputView target) throws IOException {
timestampSerializer.serialize(computationState.getTimestamp(), target);
versionSerializer.serialize(computationState.getVersion(), target);
timestampSerializer.serialize(computationState.getStartTimestamp(), target);
+ target.writeInt(computationState.getCounter());
— End diff –

How does the modification to the serialization affect backwards compatibility?

Github user kl0u commented on a diff in the pull request:

https://github.com/apache/flink/pull/3945#discussion_r117477471

— Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java —
@@ -930,6 +930,7 @@ public void serialize(NFA<T> record, DataOutputView target) throws IOException {
timestampSerializer.serialize(computationState.getTimestamp(), target);
versionSerializer.serialize(computationState.getVersion(), target);
timestampSerializer.serialize(computationState.getStartTimestamp(), target);
+ target.writeInt(computationState.getCounter());
— End diff –

The `counter` field was introduced in Flink 1.3 and the new serializer is only used in Flink 1.3. Before it was Java Serialization, so if you see in the `readComputationState()` we create a computation state with 0.

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/3945