KAFKA-3522: Add internal RecordConverter interface by mjsax · Pull Request #6150 · apache/kafka

mjsax · 2019-01-15T23:14:38Z

Part of KIP-258.

Adding the proposed RecordConverter interface.

The RecordConverter is responsible to convert ConsumerRecords from a changelog topic into <byte[], byte[]> key-value-pairs that are put into the stores. This is required to move the timestamp from the ConsumerRecord into the value-part in the store. The default implementation only maps key-value to key-value, ie, it's a no-op and this PR does not change any behavior.

mjsax · 2019-01-15T23:14:53Z

Call for review @guozhangwang @bbejeck @vvcephei

mjsax · 2019-01-15T23:16:07Z

+            final StateStore stateStore =
+                store instanceof WrappedStateStore ? ((WrappedStateStore) store).inner() : store;
+            final RecordConverter recordConverter =
+                stateStore instanceof RecordConverter ? (RecordConverter) stateStore : new DefaultRecordConverter();


Stores can implement the RecordConverter interface -- if they don't do, we use default no-op converter that maps from key-value to key-value.

mjsax · 2019-01-15T23:16:19Z

+            final StateStore stateStore =
+                store instanceof WrappedStateStore ? ((WrappedStateStore) store).inner() : store;
+            final RecordConverter recordConverter =
+                stateStore instanceof RecordConverter ? (RecordConverter) stateStore : new DefaultRecordConverter();


Stores can implement the RecordConverter interface -- if they don't do, we use default no-op converter that maps from key-value to key-value.

mjsax · 2019-01-15T23:16:49Z

+/**
+ * {@code RecordConverter} translates a {@link ConsumerRecord} into a {@link KeyValue} pair.
+ */
+// TODO: move this class to public API


After the KIP is accepted, this interface will become part of public API.

bbejeck

Thanks for the PR @mjsax, just one minor comment

bbejeck · 2019-01-16T19:34:44Z

            log.trace("Restoring state store {} from changelog topic {}", storeName, topic);
+
+            final StateStore stateStore =
+                store instanceof WrappedStateStore ? ((WrappedStateStore) store).inner() : store;


We should update the tests for setting the stateStore and recordConverter

I was able to add a test for GlobalStateManagerImpl -- also discovered a gap while trying to add a test for ProcessorStateManagerImpl: updating standby tasks was not covered by the converter -- added a test for this case.

However, I am not sure how I can test the converter for ProcessorStateManagerImpl restore case, because the ProcessorStateManagerImpl does not use the converter, but by the TaskManager -- however, it would be weird to test test in the TaskManagerTest because it has nothing to do with the task manager.

Thoughts?

I'll add some upgrade integration tests later, that will cover this case. Not sure if this would be sufficient?

\cc @guozhangwang

bbejeck · 2019-01-16T19:41:24Z

test failures kafka.server.DynamicBrokerReconfigurationTest.testUncleanLeaderElectionEnable and kafka.api.UserQuotaTest.testQuotaOverrideDelete unrelated.

retest this please

guozhangwang

Two meta comments:

could we run the simple benchmark and see if there's any significant perf impact?
Could we add more test case to ProcessorStateManager / GlobalStateManagerTest to verify that the default / customized converter is used?

mjsax · 2019-01-18T02:59:00Z

    public void shouldThrowIllegalArgumentExceptionIfAttemptingToRegisterStoreTwice() {
        stateManager.initialize();
-        initializeConsumer(2, 1, t1);
+        initializeConsumer(2, 0, t1);


This is "fix" on the side -- initializeConsumer did some weird stuff.

mjsax · 2019-01-18T03:00:01Z

Updated this.

mjsax · 2019-01-18T18:50:12Z

java.lang.AssertionError: expected:<6> but was:<5>
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.failNotEquals(Assert.java:834)
	at org.junit.Assert.assertEquals(Assert.java:645)
	at org.junit.Assert.assertEquals(Assert.java:631)
	at org.apache.kafka.streams.KafkaStreamsTest.testStateOneThreadDeadButRebalanceFinish(KafkaStreamsTest.java:211)

This test passed locally. Might be a race condition. We should monitor this. \cc @guozhangwang @bbejeck @vvcephei @ableegoldman

Retest this please

guozhangwang

LGTM. If you've verified no observable perf difference please feel free to merge

mjsax · 2019-01-18T19:52:13Z

Triggered

https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2286/ (this PR)
and
https://jenkins.confluent.io/job/system-test-kafka-branch-builder/2287/ (trunk)

Let's see :)

mjsax · 2019-01-18T23:54:08Z

Test failures in core. Retest this please.

mjsax · 2019-01-19T06:01:33Z

Move RecordConverter to public API -- will merge this after Jenkins is green. Vote passed and is closed.

mjsax · 2019-01-19T17:41:44Z

Retest this please

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

* ak/trunk: MINOR: fix race condition in KafkaStreamsTest (apache#6185) KAFKA-4850: Enable bloomfilters (apache#6012) MINOR: ducker-ak: add down -f, avoid using a terminal in ducker test KAFKA-5117: Stop resolving externalized configs in Connect REST API MINOR: Cleanup handling of mixed transactional/idempotent records (apache#6172) KAFKA-7844: Use regular subproject for generator to fix *All targets (apache#6182) Fix Documentation for cleanup.policy is out of date (apache#6181) MINOR: increase timeouts for KafkaStreamsTest (apache#6178) MINOR: Rejoin split ssl principal mapping rules (apache#6099) MINOR: Handle case where connector status endpoints returns 404 (apache#6176) MINOR: Remove unused imports, exceptions, and values (apache#6117) KAFKA-3522: Add internal RecordConverter interface (apache#6150) Fix Javadoc of KafkaConsumer (apache#6155) KAFKA-6455: Extend CacheFlushListener to forward timestamp (apache#6147) MINOR: Log partition info when creating new request batch in controller (apache#6145) KAFKA-7652: Part I; Fix SessionStore's findSession(single-key) (apache#6134) MINOR: Remove the InvalidTopicException handling in InternalTopicManager (apache#6167) [KAFKA-7024] Rocksdb state directory should be created before opening the DB (apache#6138) MINOR:: Fix typos (apache#6079)

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

KAFKA-3522: Add internal RecordConverter interface

61fbb6c

mjsax added the streams label Jan 15, 2019

mjsax commented Jan 15, 2019

View reviewed changes

bbejeck reviewed Jan 16, 2019

View reviewed changes

guozhangwang reviewed Jan 17, 2019

View reviewed changes

Github comments

75e491d

mjsax commented Jan 18, 2019

View reviewed changes

guozhangwang approved these changes Jan 18, 2019

View reviewed changes

Move RecordConvert to public API

f243566

mjsax mentioned this pull request Jan 19, 2019

KAFKA-3522: Add public interfaces for timestamped stores #6175

Merged

mjsax merged commit ed7b67d into apache:trunk Jan 20, 2019

mjsax deleted the kafka-3522-rocksdb-recordConverter branch January 20, 2019 06:54

abbccdda pushed a commit to abbccdda/kafka that referenced this pull request Jan 24, 2019

KAFKA-3522: Add internal RecordConverter interface (apache#6150)

3cb6948

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

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

KAFKA-3522: Add internal RecordConverter interface (apache#6150)

28eaf0d

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

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

Conversation

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

Uh oh!

Uh oh!

mjsax commented Jan 15, 2019

Uh oh!

mjsax Jan 15, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 15, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 15, 2019

Choose a reason for hiding this comment

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

bbejeck Jan 16, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 18, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 18, 2019

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Jan 16, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 18, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax commented Jan 18, 2019

Uh oh!

mjsax commented Jan 18, 2019

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

mjsax commented Jan 18, 2019

Uh oh!

mjsax commented Jan 18, 2019

Uh oh!

mjsax commented Jan 19, 2019

Uh oh!

mjsax commented Jan 19, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mjsax commented Jan 15, 2019 •

edited

Loading