You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2018/11/22 15:20:11 UTC
[flink] branch release-1.5 updated: [FLINK-10674] [table] Fix
handling of retractions after clean up
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.5
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.5 by this push:
new 5a533e2 [FLINK-10674] [table] Fix handling of retractions after clean up
5a533e2 is described below
commit 5a533e227254d0142181cb1b231da4a9c3d09dfe
Author: Timo Walther <tw...@apache.org>
AuthorDate: Fri Nov 9 11:19:53 2018 +0100
[FLINK-10674] [table] Fix handling of retractions after clean up
Because state clean up happens in processing time, it might be
the case that retractions are arriving after the state has
been cleaned up. Before these changes, a new accumulator was
created and invalid retraction messages were emitted. This
change drops retraction messages for which no accumulator
exists.
Backport of #7147.
---
.../table/runtime/aggregate/GroupAggProcessFunction.scala | 6 ++++++
...indowHarnessTest.scala => GroupAggregateHarnessTest.scala} | 10 +++++++---
.../apache/flink/table/runtime/harness/HarnessTestBase.scala | 11 ++++++-----
3 files changed, 19 insertions(+), 8 deletions(-)
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/GroupAggProcessFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/GroupAggProcessFunction.scala
index 3970320..f591c4f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/GroupAggProcessFunction.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/GroupAggProcessFunction.scala
@@ -95,6 +95,12 @@ class GroupAggProcessFunction(
var inputCnt = cntState.value()
if (null == accumulators) {
+ // don't create a new accumulator for unknown retractions
+ // e.g. retractions that come in right after state clean up
+ if (!inputC.change) {
+ return
+ }
+ // first accumulate message
firstRow = true
accumulators = function.createAccumulators()
} else {
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/NonWindowHarnessTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/GroupAggregateHarnessTest.scala
similarity index 96%
rename from flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/NonWindowHarnessTest.scala
rename to flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/GroupAggregateHarnessTest.scala
index 5c31cb2..fad0bd9 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/NonWindowHarnessTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/GroupAggregateHarnessTest.scala
@@ -31,13 +31,13 @@ import org.apache.flink.table.runtime.types.CRow
import org.apache.flink.types.Row
import org.junit.Test
-class NonWindowHarnessTest extends HarnessTestBase {
+class GroupAggregateHarnessTest extends HarnessTestBase {
protected var queryConfig =
new StreamQueryConfig().withIdleStateRetentionTime(Time.seconds(2), Time.seconds(3))
@Test
- def testNonWindow(): Unit = {
+ def testAggregate(): Unit = {
val processFunction = new LegacyKeyedProcessOperator[String, CRow, CRow](
new GroupAggProcessFunction(
@@ -97,7 +97,7 @@ class NonWindowHarnessTest extends HarnessTestBase {
}
@Test
- def testNonWindowWithRetract(): Unit = {
+ def testAggregateWithRetract(): Unit = {
val processFunction = new LegacyKeyedProcessOperator[String, CRow, CRow](
new GroupAggProcessFunction(
@@ -124,6 +124,10 @@ class NonWindowHarnessTest extends HarnessTestBase {
// trigger cleanup timer and register cleanup timer with 6002
testHarness.setProcessingTime(3002)
+
+ // retract after clean up
+ testHarness.processElement(new StreamRecord(CRow(Row.of(4L: JLong, 3: JInt, "ccc"), false), 4))
+
testHarness.processElement(new StreamRecord(CRow(Row.of(5L: JLong, 4: JInt, "aaa"), true), 5))
testHarness.processElement(new StreamRecord(CRow(Row.of(6L: JLong, 2: JInt, "bbb"), true), 6))
testHarness.processElement(new StreamRecord(CRow(Row.of(7L: JLong, 5: JInt, "aaa"), true), 7))
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
index 942846c..2396032 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
@@ -243,6 +243,12 @@ class HarnessTestBase {
| public final void retract(
| org.apache.flink.types.Row accs,
| org.apache.flink.types.Row input) {
+ |
+ | sum.retract(
+ | ((org.apache.flink.table.functions.aggfunctions.SumWithRetractAccumulator) accs
+ | .getField
+ | (0)),
+ | (java.lang.Integer) input.getField(1));
| }
|
| public final org.apache.flink.types.Row createAccumulators()
@@ -268,11 +274,6 @@ class HarnessTestBase {
| input.getField(0));
| }
|
- | public final void setConstantFlags(org.apache.flink.types.Row output)
- | {
- |
- | }
- |
| public final org.apache.flink.types.Row createOutputRow() {
| return new org.apache.flink.types.Row(2);
| }