[flink] branch release-1.6 updated: [FLINK-10151][state] Fix false recursion call for state TTL in TransformingStateTableKeyGroupPartitioner.tryAddToSource

[srichter]

This is an automated email from the ASF dual-hosted git repository.

srichter pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.6 by this push:
     new 00a2e81  [FLINK-10151][state] Fix false recursion call for state TTL 
in TransformingStateTableKeyGroupPartitioner.tryAddToSource
00a2e81 is described below

commit 00a2e81fc2065c599081b974cb6ce30ed1bd9ae2
Author: Andrey Zagrebin <azagre...@gmail.com>
AuthorDate: Wed Aug 15 17:20:19 2018 +0200

    [FLINK-10151][state] Fix false recursion call for state TTL in 
TransformingStateTableKeyGroupPartitioner.tryAddToSource
    
    This closes #6563.
---
 .../flink/runtime/state/heap/CopyOnWriteStateTableSnapshot.java     | 2 +-
 .../{HeapTtlStateTest.java => HeapAsyncSnapshotTtlStateTest.java}   | 6 +++---
 .../{HeapTtlStateTest.java => HeapSyncSnapshotTtlStateTest.java}    | 4 ++--
 3 files changed, 6 insertions(+), 6 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateTableSnapshot.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateTableSnapshot.java
index 11a23bf..21abf8d 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateTableSnapshot.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateTableSnapshot.java
@@ -268,7 +268,7 @@ public class CopyOnWriteStateTableSnapshot<K, N, S>
                int tryAddToSource(int currentIndex, 
CopyOnWriteStateTable.StateTableEntry<K, N, S> entry) {
                        CopyOnWriteStateTable.StateTableEntry<K, N, S> 
filteredEntry = filterEntry(entry);
                        if (filteredEntry != null) {
-                               return tryAddToSource(currentIndex, 
filteredEntry);
+                               return super.tryAddToSource(currentIndex, 
filteredEntry);
                        }
                        return currentIndex;
                }
diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapTtlStateTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapAsyncSnapshotTtlStateTest.java
similarity index 87%
copy from 
flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapTtlStateTest.java
copy to 
flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapAsyncSnapshotTtlStateTest.java
index 06de4be..02e09cf 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapTtlStateTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapAsyncSnapshotTtlStateTest.java
@@ -21,14 +21,14 @@ package org.apache.flink.runtime.state.ttl;
 import org.apache.flink.runtime.state.StateBackend;
 import org.apache.flink.runtime.state.memory.MemoryStateBackend;
 
-/** Test suite for heap state TTL. */
-public class HeapTtlStateTest extends TtlStateTestBase {
+/** Test suite for heap state TTL with asynchronous snapshotting. */
+public class HeapAsyncSnapshotTtlStateTest extends TtlStateTestBase {
        @Override
        protected StateBackendTestContext 
createStateBackendTestContext(TtlTimeProvider timeProvider) {
                return new StateBackendTestContext(timeProvider) {
                        @Override
                        protected StateBackend createStateBackend() {
-                               return new MemoryStateBackend(false);
+                               return new MemoryStateBackend(true);
                        }
                };
        }
diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapTtlStateTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapSyncSnapshotTtlStateTest.java
similarity index 90%
rename from 
flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapTtlStateTest.java
rename to 
flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapSyncSnapshotTtlStateTest.java
index 06de4be..c2c0cd2 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapTtlStateTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/HeapSyncSnapshotTtlStateTest.java
@@ -21,8 +21,8 @@ package org.apache.flink.runtime.state.ttl;
 import org.apache.flink.runtime.state.StateBackend;
 import org.apache.flink.runtime.state.memory.MemoryStateBackend;
 
-/** Test suite for heap state TTL. */
-public class HeapTtlStateTest extends TtlStateTestBase {
+/** Test suite for heap state TTL with synchronous snapshotting. */
+public class HeapSyncSnapshotTtlStateTest extends TtlStateTestBase {
        @Override
        protected StateBackendTestContext 
createStateBackendTestContext(TtlTimeProvider timeProvider) {
                return new StateBackendTestContext(timeProvider) {