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:09:00 UTC
[flink] branch release-1.6 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.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 d01dbe3 [FLINK-10674] [table] Fix handling of retractions after clean up
d01dbe3 is described below
commit d01dbe3a2adb042957355f4d80a77126412fd52d
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} | 11 +++++++----
.../apache/flink/table/runtime/harness/HarnessTestBase.scala | 11 ++++++-----
3 files changed, 19 insertions(+), 9 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 2f4bbfa..21e4dca 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
@@ -24,20 +24,19 @@ import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.streaming.api.operators.LegacyKeyedProcessOperator
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
-import org.apache.flink.table.api.StreamQueryConfig
import org.apache.flink.table.runtime.aggregate._
import org.apache.flink.table.runtime.harness.HarnessTestBase._
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 TestStreamQueryConfig(Time.seconds(2), Time.seconds(3))
@Test
- def testNonWindow(): Unit = {
+ def testAggregate(): Unit = {
val processFunction = new LegacyKeyedProcessOperator[String, CRow, CRow](
new GroupAggProcessFunction(
@@ -97,7 +96,7 @@ class NonWindowHarnessTest extends HarnessTestBase {
}
@Test
- def testNonWindowWithRetract(): Unit = {
+ def testAggregateWithRetract(): Unit = {
val processFunction = new LegacyKeyedProcessOperator[String, CRow, CRow](
new GroupAggProcessFunction(
@@ -124,6 +123,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 58fe9d3..b08f0e3 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
@@ -245,6 +245,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()
@@ -270,11 +276,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);
| }