[flink] branch master updated: [FLINK-11167] Optimize RocksDBListState#put by removing the clear before every put operation

[srichter]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 03faeb9  [FLINK-11167] Optimize RocksDBListState#put by removing the 
clear before every put operation
03faeb9 is described below

commit 03faeb9bfbc388666cf62dc2e972ee21c1370031
Author: Congxian Qiu <qcx978132...@gmail.com>
AuthorDate: Mon Apr 29 16:15:49 2019 +0800

    [FLINK-11167] Optimize RocksDBListState#put by removing the clear before 
every put operation
    
    This closes #7421.
    
    Differential Revision: https://aone.alibaba-inc.com/code/D816268
---
 .../org/apache/flink/contrib/streaming/state/RocksDBListState.java    | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git 
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
 
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
index c18adb1..03a68b2 100644
--- 
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
+++ 
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
@@ -209,8 +209,6 @@ class RocksDBListState<K, N, V>
        public void updateInternal(List<V> values) {
                Preconditions.checkNotNull(values, "List of values to add 
cannot be null.");
 
-               clear();
-
                if (!values.isEmpty()) {
                        try {
                                backend.db.put(
@@ -221,6 +219,8 @@ class RocksDBListState<K, N, V>
                        } catch (IOException | RocksDBException e) {
                                throw new FlinkRuntimeException("Error while 
updating data to RocksDB", e);
                        }
+               } else {
+                       clear();
                }
        }