fanrui created FLINK-18503:
------------------------------
Summary: bug occurs when `HeapReducingState#add` method handles
null
Key: FLINK-18503
URL: https://issues.apache.org/jira/browse/FLINK-18503
Project: Flink
Issue Type: Bug
Components: Runtime / State Backends
Affects Versions: 1.12.0
Reporter: fanrui
Fix For: 1.12.0
Attachments: image-2020-07-07-02-20-03-420.png,
image-2020-07-07-02-20-57-299.png
In our production environment, there are advertising billing jobs, which are
keyBy according to advertiserId. Calculate the cost of each advertiser in the
specified window, that is, use ReducingFunction to sum the price after keyBy by
advertiser.
But it is found that the results calculated using FsStateBackend and
RocksDBStateBackend are different. The calculation result of FsStateBackend is
wrong, and the calculation result of RocksDBStateBackend is correct.
After reading the source code, HeapReducingState#add code：[code
link|https://github.com/apache/flink/blob/f730e16fb47b0fcace7d3a1a8c8e3cb2c837ceec/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapReducingState.java#L93]
{code:java}
public void add(V value) throws IOException {
if (value == null) {
clear();
return;
}
try {
stateTable.transform(currentNamespace, value, reduceTransformation);
} catch (Exception e) {
throw new IOException("Exception while applying ReduceFunction in
reducing state", e);
}
}
{code}
If value==null, the clear method deletes the data of the current
<key,namespace> from the StateTable. ReducingFunction will only be executed if
value!=null.
h2. Why is there a bug?
For a job that calculates cost, if price != null, the price is added to result;
if price == null, then result is unchanged.
The ResultFunction method handles the case of price == null, our ResultFunction
is as follows:
{code:java}
ReduceFunction<Long> sumFunction = new ReduceFunction<Long>() {
@Override
public Long reduce(Long previousState, Long newValue) throws Exception {
// if newValue ==null,
// consider newValue to be 0 and return previousState directly
if (newValue == null) {
return previousState;
}
return previousState + newValue;
}
};
{code}
However, when HeapReducingState#add finds that the input value == null, it
directly executes the clear method, and does not execute the user-defined
ResultFunction at all.
For example: if the input prices are 17, null, and 11, the price saved in the
state is 17 when you enter 17, the price is cleared when you enter null, and
the price is 11 when you enter 11, so the result is wrong.
Fortunately, the calculation result of RocksDBStateBackend is correct. The
RocksDBReducingState#add method does not perform special treatment for null.
RocksDBReducingState#add code is as follows:[code
link|https://github.com/apache/flink/blob/f730e16fb47b0fcace7d3a1a8c8e3cb2c837ceec/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java#L92]
{code:java}
public void add(V value) throws Exception {
byte[] key = getKeyBytes();
V oldValue = getInternal(key);
V newValue = oldValue == null ? value : reduceFunction.reduce(oldValue,
value);
updateInternal(key, newValue);
}
{code}
h2. Flink UT can reproduce this bug
StateBackendTestBase#testReducingStateAddAndGet can reproduce this bug.
Need to be modified as follows：
# udf
{code:java}
ReduceFunction<Long> sumFunction = new ReduceFunction<Long>() {
@Override
public Long reduce(Long previousState, Long newValue) throws Exception {
// if newValue ==null,
// consider newValue to be 0 and return previousState directly
if (newValue == null) {
return previousState;
}
return previousState + newValue;
}
};
final ReducingStateDescriptor<Long> stateDescr =
new ReducingStateDescriptor<>("my-state", sumFunction, Long.class);{code}
# add element
{code:java}
keyedBackend.setCurrentKey("def");
assertNull(state.get());
state.add(17L);
state.add(null);//new code
state.add(11L);
assertEquals(28L, state.get().longValue());{code}
My code repository commit
[link|https://github.com/1996fanrui/flink/commit/645118dd2f95de88580d07e00d88e8783a0f9680]
The UT execution output of RocksDBStateBackendTest is as follows:
!image-2020-07-07-02-20-03-420.png!
The UT execution output of FileStateBackendTest&MemoryStateBackendTest is as
follows:
!image-2020-07-07-02-20-57-299.png!
{code:java}
java.lang.AssertionError:
Expected :28
Actual :11{code}
The above phenomenon shows that the HeapReducingState#add method has a bug.
Regardless of which state backend you choose, the semantics provided by the
Flink engine should be consistent and should not output different calculation
results.
h2. My solution
Remove the processing logic of value == null in HeapReducingState#add. Result:
All UTs of FileStateBackendTest can be passed.
h2. Similar bug
HeapFoldingState#add & HeapAggregatingState#add
h2. Question
HeapReducingState#add When designing, why does the designer handle the null
case specially? I think the case of null should be handled by the user-defined
ReducingFunction.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
