[jira] [Created] (FLINK-36897) Error in calling processElement for AbstractAsynchronous StateStreamOperator

["Wang Qilong (Jira)" <ji...@apache.org>]

Wang Qilong created FLINK-36897:
-----------------------------------

             Summary: Error in calling processElement for AbstractAsynchronous StateStreamOperator
                 Key: FLINK-36897
                 URL: https://issues.apache.org/jira/browse/FLINK-36897
             Project: Flink
          Issue Type: Bug
          Components: Table SQL / Runtime
    Affects Versions: 2.0.0
            Reporter: Wang Qilong


When I created the AbstractAsynchronous StateMapBundleOperator and inherited it from the AbstractAsynchronous StateStreamOperator, there was an error in the data passed into the element by the processElement of the AbstractAsynchronous StateMapBundleOperator itself

The inheritance relationship between asynchronous synchronization and two classes is:
AbstractMapBundleOperator->AbstractStreamOperator
AbstractAsyncStateMapBundleOperator->AbstractAsyncStateStreamOperator->AbstractStreamOperator
The reason for creating this class is to enable KeyedMapBundleOperator to support asynchronous running capability

Example of incorrect information: For example, the original data format was:
val data = new mutable.MutableList[(String, Long)]
data.+=(("x", 1L))
data.+=(("x", 2L))
data.+=(("x", 3L))
data.+=(("y", 1L))
data.+=(("y", 2L))
data.+=(("z", 3L))
So the result of data transmission becomes:
val data = new mutable.MutableList[(String, Long)]
data.+=(("x", 1L))
data.+=(("x", 2L))
data.+=(("x", 3L))
data.+=(("x", 1L))
data.+=(("x", 2L))
data.+=(("x", 3L))

How to reproduce:
Run testOverloadedAccumulator in SQL/AggregateITCase.jva in [1]

 

[1]  https://github.com/Au-Miner/flink/tree/FLINK-36882



--
This message was sent by Atlassian Jira
(v8.20.10#820010)