You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ca...@apache.org on 2023/12/05 09:02:14 UTC
(kafka) branch trunk updated: KAFKA-14412: Better Rocks column family management (#14852)
This is an automated email from the ASF dual-hosted git repository.
cadonna pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 20a223061c6 KAFKA-14412: Better Rocks column family management (#14852)
20a223061c6 is described below
commit 20a223061c6cbb6ebae7ddfacc4bc89f807bda70
Author: Nick Telford <ni...@gmail.com>
AuthorDate: Tue Dec 5 09:02:04 2023 +0000
KAFKA-14412: Better Rocks column family management (#14852)
When opening RocksDB, we were checking for an error in
`RocksDBTimestampedStore` to detect if the `keyValueWithTimestamp` CF is
missing.
The `openRocksDB` method now supports any number of column families, not
just the extra one used by `RocksDBTimestampedStore`. We now check for
the existing column families _before_ opening the database, which allows
us to create any missing column families.
Supporting automatic creation of any number of missing column families
is a pre-requisite for KIP-892: Transactional StateStores.
Reviewers: Anna Sophie Blee-Goldman <ab...@apache.org>, Bruno Cadonna <ca...@apache.org>
---
.../streams/state/internals/RocksDBStore.java | 71 ++++++++++++++++++++--
.../state/internals/RocksDBTimestampedStore.java | 36 +++--------
2 files changed, 75 insertions(+), 32 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 d21279a2fd9..77e0c1e05e3 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
@@ -66,13 +66,16 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.stream.Collectors;
import static org.apache.kafka.streams.StreamsConfig.InternalConfig.IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED;
import static org.apache.kafka.streams.StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG;
@@ -278,18 +281,76 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
void openRocksDB(final DBOptions dbOptions,
final ColumnFamilyOptions columnFamilyOptions) {
- final List<ColumnFamilyDescriptor> columnFamilyDescriptors
- = Collections.singletonList(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions));
- final List<ColumnFamilyHandle> columnFamilies = new ArrayList<>(columnFamilyDescriptors.size());
+ final List<ColumnFamilyHandle> columnFamilies = openRocksDB(
+ dbOptions,
+ new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions)
+ );
+
+ dbAccessor = new SingleColumnFamilyAccessor(columnFamilies.get(0));
+ }
+
+ /**
+ * Open RocksDB while automatically creating any requested column families that don't yet exist.
+ */
+ protected List<ColumnFamilyHandle> openRocksDB(final DBOptions dbOptions,
+ final ColumnFamilyDescriptor defaultColumnFamilyDescriptor,
+ final ColumnFamilyDescriptor... columnFamilyDescriptors) {
+ final String absolutePath = dbDir.getAbsolutePath();
+ final List<ColumnFamilyDescriptor> extraDescriptors = Arrays.asList(columnFamilyDescriptors);
+ final List<ColumnFamilyDescriptor> allDescriptors = new ArrayList<>(1 + columnFamilyDescriptors.length);
+ allDescriptors.add(defaultColumnFamilyDescriptor);
+ allDescriptors.addAll(extraDescriptors);
try {
- db = RocksDB.open(dbOptions, dbDir.getAbsolutePath(), columnFamilyDescriptors, columnFamilies);
- dbAccessor = new SingleColumnFamilyAccessor(columnFamilies.get(0));
+ final Options options = new Options(dbOptions, defaultColumnFamilyDescriptor.getOptions());
+ final List<byte[]> allExisting = RocksDB.listColumnFamilies(options, absolutePath);
+
+ final List<ColumnFamilyDescriptor> existingDescriptors = new LinkedList<>();
+ existingDescriptors.add(defaultColumnFamilyDescriptor);
+ existingDescriptors.addAll(extraDescriptors.stream()
+ .filter(descriptor -> allExisting.stream().anyMatch(existing -> Arrays.equals(existing, descriptor.getName())))
+ .collect(Collectors.toList()));
+ final List<ColumnFamilyDescriptor> toCreate = extraDescriptors.stream()
+ .filter(descriptor -> allExisting.stream().noneMatch(existing -> Arrays.equals(existing, descriptor.getName())))
+ .collect(Collectors.toList());
+ final List<ColumnFamilyHandle> existingColumnFamilies = new ArrayList<>(existingDescriptors.size());
+ db = RocksDB.open(dbOptions, absolutePath, existingDescriptors, existingColumnFamilies);
+ final List<ColumnFamilyHandle> createdColumnFamilies = db.createColumnFamilies(toCreate);
+
+ return mergeColumnFamilyHandleLists(existingColumnFamilies, createdColumnFamilies, allDescriptors);
+
} catch (final RocksDBException e) {
throw new ProcessorStateException("Error opening store " + name + " at location " + dbDir.toString(), e);
}
}
+ /**
+ * match up the existing and created ColumnFamilyHandles with the existing/created ColumnFamilyDescriptors
+ * so that the order of the resultant List matches the order of the allDescriptors argument
+ */
+ private List<ColumnFamilyHandle> mergeColumnFamilyHandleLists(final List<ColumnFamilyHandle> existingColumnFamilyHandles,
+ final List<ColumnFamilyHandle> createdColumnFamilyHandles,
+ final List<ColumnFamilyDescriptor> allDescriptors) throws RocksDBException {
+ final List<ColumnFamilyHandle> columnFamilies = new ArrayList<>(allDescriptors.size());
+ int existing = 0;
+ int created = 0;
+
+ while (existing + created < allDescriptors.size()) {
+ final ColumnFamilyHandle existingHandle = existing < existingColumnFamilyHandles.size() ? existingColumnFamilyHandles.get(existing) : null;
+ final ColumnFamilyHandle createdHandle = created < createdColumnFamilyHandles.size() ? createdColumnFamilyHandles.get(created) : null;
+ if (existingHandle != null && Arrays.equals(existingHandle.getDescriptor().getName(), allDescriptors.get(existing + created).getName())) {
+ columnFamilies.add(existingHandle);
+ existing++;
+ } else if (createdHandle != null && Arrays.equals(createdHandle.getDescriptor().getName(), allDescriptors.get(existing + created).getName())) {
+ columnFamilies.add(createdHandle);
+ created++;
+ } else {
+ throw new IllegalStateException("Unable to match up column family handles with descriptors.");
+ }
+ }
+ return columnFamilies;
+ }
+
@Override
public String name() {
return name;
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java
index ea940ca8fb6..7ef8d1ba06c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java
@@ -35,13 +35,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Objects;
-import static java.util.Arrays.asList;
import static org.apache.kafka.streams.state.TimestampedBytesStore.convertToTimestampedFormat;
/**
@@ -50,6 +48,8 @@ import static org.apache.kafka.streams.state.TimestampedBytesStore.convertToTime
public class RocksDBTimestampedStore extends RocksDBStore implements TimestampedBytesStore {
private static final Logger log = LoggerFactory.getLogger(RocksDBTimestampedStore.class);
+ private static final byte[] TIMESTAMPED_VALUES_COLUMN_FAMILY_NAME = "keyValueWithTimestamp".getBytes(StandardCharsets.UTF_8);
+
public RocksDBTimestampedStore(final String name,
final String metricsScope) {
super(name, metricsScope);
@@ -64,31 +64,14 @@ public class RocksDBTimestampedStore extends RocksDBStore implements Timestamped
@Override
void openRocksDB(final DBOptions dbOptions,
final ColumnFamilyOptions columnFamilyOptions) {
- final List<ColumnFamilyDescriptor> columnFamilyDescriptors = asList(
- new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
- new ColumnFamilyDescriptor("keyValueWithTimestamp".getBytes(StandardCharsets.UTF_8), columnFamilyOptions));
- final List<ColumnFamilyHandle> columnFamilies = new ArrayList<>(columnFamilyDescriptors.size());
-
- try {
- db = RocksDB.open(dbOptions, dbDir.getAbsolutePath(), columnFamilyDescriptors, columnFamilies);
- setDbAccessor(columnFamilies.get(0), columnFamilies.get(1));
- } catch (final RocksDBException e) {
- if ("Column family not found: keyValueWithTimestamp".equals(e.getMessage())) {
- try {
- db = RocksDB.open(dbOptions, dbDir.getAbsolutePath(), columnFamilyDescriptors.subList(0, 1), columnFamilies);
- columnFamilies.add(db.createColumnFamily(columnFamilyDescriptors.get(1)));
- } catch (final RocksDBException fatal) {
- throw new ProcessorStateException("Error opening store " + name + " at location " + dbDir.toString(), fatal);
- }
- setDbAccessor(columnFamilies.get(0), columnFamilies.get(1));
- } else {
- throw new ProcessorStateException("Error opening store " + name + " at location " + dbDir.toString(), e);
- }
- }
- }
+ final List<ColumnFamilyHandle> columnFamilies = openRocksDB(
+ dbOptions,
+ new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
+ new ColumnFamilyDescriptor(TIMESTAMPED_VALUES_COLUMN_FAMILY_NAME, columnFamilyOptions)
+ );
+ final ColumnFamilyHandle noTimestampColumnFamily = columnFamilies.get(0);
+ final ColumnFamilyHandle withTimestampColumnFamily = columnFamilies.get(1);
- private void setDbAccessor(final ColumnFamilyHandle noTimestampColumnFamily,
- final ColumnFamilyHandle withTimestampColumnFamily) {
final RocksIterator noTimestampsIter = db.newIterator(noTimestampColumnFamily);
noTimestampsIter.seekToFirst();
if (noTimestampsIter.isValid()) {
@@ -102,7 +85,6 @@ public class RocksDBTimestampedStore extends RocksDBStore implements Timestamped
noTimestampsIter.close();
}
-
private class DualColumnFamilyAccessor implements RocksDBAccessor {
private final ColumnFamilyHandle oldColumnFamily;
private final ColumnFamilyHandle newColumnFamily;