Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

KAFKA-8791: RocksDBTimestampedStore should open in regular mode by default #7201

Merged
merged 3 commits into from
Aug 14, 2019
Merged

KAFKA-8791: RocksDBTimestampedStore should open in regular mode by default #7201

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4a07622
KAFKA-8791: RocksDBTimestampedStore should open in regular mode by de…
mjsax Aug 13, 2019
06359e0
improve exception handling in test code
mjsax Aug 13, 2019
4aaffde
fix checkstyle
mjsax Aug 13, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 16 additions & 16 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,20 +70,7 @@ void openRocksDB(final DBOptions dbOptions,

try {
db = RocksDB.open(dbOptions, dbDir.getAbsolutePath(), columnFamilyDescriptors, columnFamilies);

final ColumnFamilyHandle noTimestampColumnFamily = columnFamilies.get(0);

final RocksIterator noTimestampsIter = db.newIterator(noTimestampColumnFamily);
noTimestampsIter.seekToFirst();
if (noTimestampsIter.isValid()) {
log.info("Opening store {} in upgrade mode", name);
dbAccessor = new DualColumnFamilyAccessor(noTimestampColumnFamily, columnFamilies.get(1));
} else {
log.info("Opening store {} in regular mode", name);
dbAccessor = new SingleColumnFamilyAccessor(columnFamilies.get(1));
noTimestampColumnFamily.close();
}
noTimestampsIter.close();
setDbAccessor(columnFamilies.get(0), columnFamilies.get(1));
} catch (final RocksDBException e) {
if ("Column family not found: : keyValueWithTimestamp".equals(e.getMessage())) {
try {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe add an INFO or WARN statement here to help with tracing control flow.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure -- it's an internal implementation detail that we create a second CF -- if not a WARN for sure, because it's expected to happen. We could de a DEBUG if we want.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good point. I'd say just leave it as is.

Expand All @@ -92,14 +79,27 @@ void openRocksDB(final DBOptions dbOptions,
} catch (final RocksDBException fatal) {
throw new ProcessorStateException("Error opening store " + name + " at location " + dbDir.toString(), fatal);
}
log.info("Opening store {} in upgrade mode", name);
dbAccessor = new DualColumnFamilyAccessor(columnFamilies.get(0), columnFamilies.get(1));
setDbAccessor(columnFamilies.get(0), columnFamilies.get(1));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Line 75 above: double colons?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes -- well, it's the error message text we get from RocksDB -- nothing we can do about it guess.

} else {
throw new ProcessorStateException("Error opening store " + name + " at location " + dbDir.toString(), e);
}
}
}

private void setDbAccessor(final ColumnFamilyHandle noTimestampColumnFamily,
final ColumnFamilyHandle withTimestampColumnFamily) {
final RocksIterator noTimestampsIter = db.newIterator(noTimestampColumnFamily);
noTimestampsIter.seekToFirst();
if (noTimestampsIter.isValid()) {
log.info("Opening store {} in upgrade mode", name);
dbAccessor = new DualColumnFamilyAccessor(noTimestampColumnFamily, withTimestampColumnFamily);
} else {
log.info("Opening store {} in regular mode", name);
dbAccessor = new SingleColumnFamilyAccessor(withTimestampColumnFamily);
noTimestampColumnFamily.close();
}
noTimestampsIter.close();
}


private class DualColumnFamilyAccessor implements RocksDBAccessor {
Expand Down
Loading