root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Revert "MINOR: make flush no-op as we don't need to call flush on commit." 

This reverts commit 90b2a2bf66.
This commit is contained in:
Guozhang Wang 2017-06-21 13:05:54 -07:00
parent 254add9534
commit f848e2cd68
1 changed files with 3 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
View File

@ -95,7 +95,6 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
private Options options; private Options options;
private WriteOptions wOptions; private WriteOptions wOptions;
private FlushOptions fOptions; private FlushOptions fOptions;
private boolean eosEnabled;
protected volatile boolean open = false; protected volatile boolean open = false;
@ -179,7 +178,7 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
putInternal(key, value); putInternal(key, value);
} }
}); });
eosEnabled = checkForEos(context.appConfigs());
open = true; open = true;
} }
@ -356,12 +355,12 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
@Override @Override
public synchronized void flush() { public synchronized void flush() {
if (db == null || eosEnabled) { if (db == null) {
return; return;
} }
// flush RocksDB
flushInternal(); flushInternal();
} }
/** /**
* @throws ProcessorStateException if flushing failed because of any internal store exceptions * @throws ProcessorStateException if flushing failed because of any internal store exceptions
*/ */
@ -373,10 +372,6 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
} }
} }
private boolean checkForEos(Map<String, Object> configs) {
return StreamsConfig.EXACTLY_ONCE.equals(configs.get(StreamsConfig.PROCESSING_GUARANTEE_CONFIG));
}
@Override @Override
public synchronized void close() { public synchronized void close() {
if (!open) { if (!open) {