KAFKA-3522: Add RocksDBTimestampedSegmentedBytesStore by mjsax · Pull Request #6186 · apache/kafka

mjsax · 2019-01-23T06:23:54Z

Part of KIP-258.

mjsax · 2019-03-13T19:55:44Z

Call for review @guozhangwang @bbejeck @vvcephei @ableegoldman

mjsax · 2019-03-13T19:56:40Z

+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.EXPIRED_WINDOW_RECORD_DROP;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
+
+public class AbstractSegmentedBytesStore<S extends Segment> implements SegmentedBytesStore {


This replaces/generalizes RocksDBSegmentedBytesStore to work with both segment types

mjsax · 2019-03-13T19:57:16Z

-import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.EXPIRED_WINDOW_RECORD_DROP;
-import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
-
-public class RocksDBSegmentedBytesStore implements SegmentedBytesStore {


All logic is moved into AbstractSegmentedBytesStore.

bbejeck

I've made a pass and overall this looks good to me. I only have some comments regarding testing.
Looks like TimestampedSegments and TimestampedSegment need unit tests.

Also with the StoresTest do we want to add a test ensuring that the underlying SegmentedBytesStore is of the correct type (not sure how hard this will be)? I add this only because it will help now and in the near future when we switch to have KIP-258 enabled.

vvcephei

Hey, thanks for the PR, @mjsax !

I've left a few comments.

vvcephei · 2019-03-14T17:07:07Z

+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteBatch;


The name of this class implies that it's abstracting over all segmented bytes stores, but it depends on RocksDB... Should we rename the class to AbstractRocksDBSegmentedBytesStore? (Or abstract away the concept of a batch?)

vvcephei · 2019-03-14T17:11:15Z

+import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteBatch;
+
+public interface BulkLoadingKeyValueStore extends KeyValueStore<Bytes, byte[]> {


It seems unnecessary for this interface to subtype KeyValueStore<Bytes, byte[]>. The methods it defines are unrelated to the key/value store.

Should this just be an independent interface to be mixed in to classes that support bulk loading? It could be named BulkLoadingStore.

vvcephei · 2019-03-14T17:15:08Z

+                                          final long retention,
+                                          final long segmentInterval,
+                                          final KeySchema keySchema) {
+        super(name, metricScope, keySchema, new TimestampedSegments(name, retention, segmentInterval));


Instead of declaring two subclasses that only function to pass different arguments to the constructor of their superclass, can we just inline these constructors and pass the desired different arguments at the call site?

We could. I prefer the current way because it seems easier to use.

I'm fine with either approach, but testing coverage wise:

If we want to keep a separate class, should we also add a RocksDBTimestampedSegmentedBytesStoreTest?

If we just want to maintain a single class, say under RocksDBSegmentedBytesStore as the current AbstractRocksDBSegmentedBytesStore, then should we augment the existing test file with timestamps enabled as well?

Will add RocksDBTimestampedSegmentedBytesStoreTest.

mjsax · 2019-03-14T23:23:06Z

Updated per comments. Thanks @bbejeck and @vvcephei

mjsax · 2019-03-16T02:48:52Z

Java8 passed. Java11 timed out.

Retest this please.

mjsax · 2019-03-16T22:43:40Z

Java8 failed with https://issues.apache.org/jira/browse/KAFKA-7965

Java 11 passed.

Retest this please.

guozhangwang

I've got one meta question about the additional TimestampedSegments and TimestampedSegment classes, others are all nit.

guozhangwang · 2019-03-18T18:00:07Z

+                                          final long retention,
+                                          final long segmentInterval,
+                                          final KeySchema keySchema) {
+        super(name, metricScope, keySchema, new TimestampedSegments(name, retention, segmentInterval));


I'm fine with either approach, but testing coverage wise:

If we want to keep a separate class, should we also add a RocksDBTimestampedSegmentedBytesStoreTest?

If we just want to maintain a single class, say under RocksDBSegmentedBytesStore as the current AbstractRocksDBSegmentedBytesStore, then should we augment the existing test file with timestamps enabled as well?

guozhangwang · 2019-03-18T18:01:46Z

+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.EXPIRED_WINDOW_RECORD_DROP;
+import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
+
+public class AbstractRocksDBSegmentedBytesStore<S extends Segment> implements SegmentedBytesStore {


Just to clarify there's no logic changes from the existing RocksDBSegmentedBytesStore, but just extract it out right (if yes I'll skip reviewing this file)?

Yes. It's the same code.

guozhangwang · 2019-03-18T21:10:18Z

 import java.io.IOException;

-public interface Segment extends StateStore {
+public interface Segment extends KeyValueStore<Bytes, byte[]>, BulkLoadingStore {


We keep the extended super class intentionally so that in the future we can implement the segment differently (e.g. not based on rocksDB at all), but since this is an internal class anyways we can always change it back when needed to, so I guess this is fine --- just to leave the context on this.

guozhangwang · 2019-03-18T21:13:36Z

+import java.io.IOException;
+import java.util.Objects;
+
+class TimestampedSegment extends RocksDBTimestampedStore implements Comparable<TimestampedSegment>, Segment {


What's the benefits of keeping this and the following TimestampedSegments along from KeyValueSegment? By looking at the source code now they seem exactly the same. Is it going to be changed in follow-up PRs?

It's both quite the same, but here we extend RocksDBTimestampedStore while the other extends RocksDBStore. Because Java does not support multi-inheritance, we cannot share and inherit code but need to duplicate it here.

Do you have an idea how we could avoid this?

Fix bug revealed by new test

mjsax · 2019-03-20T06:35:13Z

+                }
+                try {
+                    final WriteBatch batch = writeBatchMap.computeIfAbsent(segment, s -> new WriteBatch());
+                    segment.addToBatch(record, batch);


This is the bug fix -- in the original code, the record was added to the batch directly (ie, to default CF) -- we know delegate this to the segment that can decide to which CF the record is added.

mjsax · 2019-03-20T06:38:05Z


+        void addToBatch(final byte[] key,
+                        final byte[] value,
+                        final WriteBatch batch) throws RocksDBException;


We add this new method to RocksDBAccessor for the fix

mjsax · 2019-03-20T06:38:48Z

-                } else {
-                    batch.put(columnFamily, entry.key.get(), entry.value);
-                }
+                addToBatch(entry.key.get(), entry.value, batch);


some refactoring to share more code exploiting the newly added method (similar below)

mjsax · 2019-03-20T06:41:03Z

+import static org.junit.Assert.assertTrue;
+
+@RunWith(Parameterized.class)
+public abstract class AbstractRocksDBSegmentedBytesStoreTest<S extends Segment> {


Refactored the test to reuse test code for both segments. It's the same test code as before, with some generic/abstract types/methods

mjsax · 2019-03-20T06:42:44Z

Updated this.

guozhangwang

After the discussion with @mjsax I think I understand the trade-offs of code savings now. I'll leave it to his call on whether to merge RocksDBXXSegmentedByteStore classes.

@mjsax please feel free to merge whenever you feel comfortable.

mjsax · 2019-03-26T21:22:18Z

Ack. It's internal and we can refactor later if it turns out to be worth it. Merging as-is for now.

* apache/trunk: KAFKA-8030: Fix flaky tests in TopicCommandWithAdminClientTest fix compile error for example (apache#6526) MINOR: Comment spelling nit MINOR: Optimize ConnectionStressWorker KAFKA-8034: Use automatic RPC generation in DeleteTopics MINOR: Move KTable source topic for changelog to optimization framework (apache#6500) KAFKA-7502: Cleanup KTable materialization logic in a single place (doMapValues) (apache#6520) Cleanup KTableImpl#doTransformValues (apache#6519) MINOR: Streams input topic corrected (apache#6513) MINOR: WorkerUtils#abort: fix bug in abort logic (apache#6516) KAFKA-7502: Cleanup KTable materialization logic in a single place (filter) (apache#6453) MINOR: Fix some spelling corrections in comments (apache#6507) KAFKA-3522: Add RocksDBTimestampedSegmentedBytesStore (apache#6186) MINOR: Add 2.2.0 upgrade instructions (apache#6501)

Reviewers: Bill Bejeck <bill@confluent.io>, John Roesler <john@confluent.io>, Guozhang Wang <guozhang@confluent.io>

mjsax added the streams label Jan 23, 2019

mjsax force-pushed the kafka-3522-rocksdb-format-rocksdbWindowWithTSStore branch from e15d7e1 to 47e3489 Compare January 27, 2019 05:34

KAFKA-3522: Add RocksDBTimestampedSegmentedBytesStore

dec3072

mjsax force-pushed the kafka-3522-rocksdb-format-rocksdbWindowWithTSStore branch from 47e3489 to dec3072 Compare March 13, 2019 18:58

mjsax changed the title ~~KAFKA-3522: Add RocksDBTimestampedSegmentedBytesStore [WIP]~~ KAFKA-3522: Add RocksDBTimestampedSegmentedBytesStore Mar 13, 2019

mjsax commented Mar 13, 2019

View reviewed changes

bbejeck reviewed Mar 14, 2019

View reviewed changes

vvcephei reviewed Mar 14, 2019

View reviewed changes

Github comments

b81e873

guozhangwang reviewed Mar 18, 2019

View reviewed changes

mjsax added 2 commits March 19, 2019 23:30

Add RocksDBTimestampedSegmentedBytesStoreTest

51d34b8

Fix bug revealed by new test

x

39578f4

mjsax commented Mar 20, 2019

View reviewed changes

guozhangwang approved these changes Mar 26, 2019

View reviewed changes

mjsax merged commit dc0601a into apache:trunk Mar 26, 2019

mjsax deleted the kafka-3522-rocksdb-format-rocksdbWindowWithTSStore branch March 26, 2019 21:23

pengxiaolong pushed a commit to pengxiaolong/kafka that referenced this pull request Jun 14, 2019

KAFKA-3522: Add RocksDBTimestampedSegmentedBytesStore (apache#6186)

0877afa

Reviewers: Bill Bejeck <bill@confluent.io>, John Roesler <john@confluent.io>, Guozhang Wang <guozhang@confluent.io>

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Conversation

mjsax commented Jan 23, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax commented Mar 13, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax commented Mar 14, 2019

Uh oh!

mjsax commented Mar 16, 2019

Uh oh!

mjsax commented Mar 16, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax Mar 20, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax commented Mar 20, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

mjsax commented Mar 26, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

mjsax commented Jan 23, 2019 •

edited

Loading

mjsax Mar 20, 2019 •

edited

Loading