kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject [kafka] branch 2.0 updated: KAFKA-7023: Move prepareForBulkLoad() call after customized RocksDBConfigSetter (#5166)
Date Tue, 12 Jun 2018 01:05:35 GMT
This is an automated email from the ASF dual-hosted git repository.

guozhang pushed a commit to branch 2.0
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.0 by this push:
     new 156a1c4  KAFKA-7023: Move prepareForBulkLoad() call after customized RocksDBConfigSetter
(#5166)
156a1c4 is described below

commit 156a1c4dffebbb2b389e75b767fa9b57943c24d6
Author: Liquan Pei <liquanpei@gmail.com>
AuthorDate: Mon Jun 11 13:32:03 2018 -0700

    KAFKA-7023: Move prepareForBulkLoad() call after customized RocksDBConfigSetter (#5166)
    
    *Summary
    options.prepareForBulkLoad() and then use the configs from the customized customized RocksDBConfigSetter.
This may overwrite the configs set in prepareBulkLoad call. The fix is to move prepareBulkLoad
call after applying configs customized RocksDBConfigSetter.
    
    Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <guozhang@confluent.io>,
Bill Bejeck <bill@confluent.io>
---
 .../org/apache/kafka/streams/state/internals/RocksDBStore.java   | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
index cfef035..6084ecb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
@@ -130,10 +130,6 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>
{
         // (this could be a bug in the RocksDB code and their devs have been contacted).
         options.setIncreaseParallelism(Math.max(Runtime.getRuntime().availableProcessors(),
2));
 
-        if (prepareForBulkload) {
-            options.prepareForBulkLoad();
-        }
-
         wOptions = new WriteOptions();
         wOptions.setDisableWAL(true);
 
@@ -148,6 +144,11 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>
{
             final RocksDBConfigSetter configSetter = Utils.newInstance(configSetterClass);
             configSetter.setConfig(name, options, configs);
         }
+
+        if (prepareForBulkload) {
+            options.prepareForBulkLoad();
+        }
+
         this.dbDir = new File(new File(context.stateDir(), parentDir), this.name);
 
         try {

-- 
To stop receiving notification emails like this one, please contact
guozhang@apache.org.

Mime
View raw message