private RocksDB initializeRocksDBWithBookieConf()

in bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDB.java [188:293]
92 lines of code
12 McCabe index (conditional complexity)

    private RocksDB initializeRocksDBWithBookieConf(String basePath, String subPath, DbConfigType dbConfigType,
                                           ServerConfiguration conf, boolean readOnly) throws IOException {
        Options options = new Options();
        options.setCreateIfMissing(true);
        ChecksumType checksumType = ChecksumType.valueOf(conf.getString(ROCKSDB_CHECKSUM_TYPE, "kxxHash"));
        int formatVersion = conf.getInt(ROCKSDB_FORMAT_VERSION, 5);

        if (dbConfigType == DbConfigType.EntryLocation) {
            /* Set default RocksDB block-cache size to 10% / numberOfLedgers of direct memory, unless override */
            int ledgerDirsSize = conf.getLedgerDirNames().length;
            long defaultRocksDBBlockCacheSizeBytes = maxDirectMemory() / ledgerDirsSize / 10;
            long blockCacheSize = DbLedgerStorage.getLongVariableOrDefault(conf, ROCKSDB_BLOCK_CACHE_SIZE,
                defaultRocksDBBlockCacheSizeBytes);

            long writeBufferSizeMB = conf.getInt(ROCKSDB_WRITE_BUFFER_SIZE_MB, 64);
            long sstSizeMB = conf.getInt(ROCKSDB_SST_SIZE_MB, 64);
            int numLevels = conf.getInt(ROCKSDB_NUM_LEVELS, -1);
            int numFilesInLevel0 = conf.getInt(ROCKSDB_NUM_FILES_IN_LEVEL0, 4);
            long maxSizeInLevel1MB = conf.getLong(ROCKSDB_MAX_SIZE_IN_LEVEL1_MB, 256);
            int blockSize = conf.getInt(ROCKSDB_BLOCK_SIZE, 64 * 1024);
            int bloomFilterBitsPerKey = conf.getInt(ROCKSDB_BLOOM_FILTERS_BITS_PER_KEY, 10);
            boolean lz4CompressionEnabled = conf.getBoolean(ROCKSDB_LZ4_COMPRESSION_ENABLED, true);

            if (lz4CompressionEnabled) {
                options.setCompressionType(CompressionType.LZ4_COMPRESSION);
            }
            options.setWriteBufferSize(writeBufferSizeMB * 1024 * 1024);
            options.setMaxWriteBufferNumber(4);
            if (numLevels > 0) {
                options.setNumLevels(numLevels);
            }
            options.setLevelZeroFileNumCompactionTrigger(numFilesInLevel0);
            options.setMaxBytesForLevelBase(maxSizeInLevel1MB * 1024 * 1024);
            options.setMaxBackgroundJobs(32);
            options.setIncreaseParallelism(32);
            options.setMaxTotalWalSize(512 * 1024 * 1024);
            options.setMaxOpenFiles(-1);
            options.setTargetFileSizeBase(sstSizeMB * 1024 * 1024);
            options.setDeleteObsoleteFilesPeriodMicros(TimeUnit.HOURS.toMicros(1));

            this.cache = new LRUCache(blockCacheSize);
            BlockBasedTableConfig tableOptions = new BlockBasedTableConfig();
            tableOptions.setBlockSize(blockSize);
            tableOptions.setBlockCache(cache);
            tableOptions.setFormatVersion(formatVersion);
            tableOptions.setChecksumType(checksumType);
            if (bloomFilterBitsPerKey > 0) {
                tableOptions.setFilterPolicy(new BloomFilter(bloomFilterBitsPerKey, false));
            }

            // Options best suited for HDDs
            tableOptions.setCacheIndexAndFilterBlocks(true);
            options.setLevelCompactionDynamicLevelBytes(true);

            options.setTableFormatConfig(tableOptions);
        } else {
            this.cache = null;
            BlockBasedTableConfig tableOptions = new BlockBasedTableConfig();
            tableOptions.setFormatVersion(formatVersion);
            tableOptions.setChecksumType(checksumType);
            options.setTableFormatConfig(tableOptions);
        }

            // Configure file path
        String logPath = conf.getString(ROCKSDB_LOG_PATH, "");
        if (!logPath.isEmpty()) {
            Path logPathSetting = FileSystems.getDefault().getPath(logPath, subPath);
            Files.createDirectories(logPathSetting);
            log.info("RocksDB<{}> log path: {}", subPath, logPathSetting);
            options.setDbLogDir(logPathSetting.toString());
        }
        this.dbPath = FileSystems.getDefault().getPath(basePath, subPath).toFile().toString();

        // Configure log level
        String logLevel = conf.getString(ROCKSDB_LOG_LEVEL, "info");
        switch (logLevel) {
            case "debug":
                options.setInfoLogLevel(InfoLogLevel.DEBUG_LEVEL);
                break;
            case "info":
                options.setInfoLogLevel(InfoLogLevel.INFO_LEVEL);
                break;
            case "warn":
                options.setInfoLogLevel(InfoLogLevel.WARN_LEVEL);
                break;
            case "error":
                options.setInfoLogLevel(InfoLogLevel.ERROR_LEVEL);
                break;
            default:
                log.warn("Unrecognized RockDB log level: {}", logLevel);
        }

            // Keep log files for 1month
        options.setKeepLogFileNum(30);
        options.setLogFileTimeToRoll(TimeUnit.DAYS.toSeconds(1));
        this.options = options;
        try {
            if (readOnly) {
                return RocksDB.openReadOnly(options, dbPath);
            } else {
                return RocksDB.open(options, dbPath);
            }
        } catch (RocksDBException e) {
            throw new IOException("Error open RocksDB database", e);
        }
    }