KAFKA-3522: Replace RecordConverter with TimestampedBytesStore by mjsax · Pull Request #6204 · apache/kafka

mjsax · 2019-01-26T10:01:17Z

This is a change to KIP-258 as announced on the mailing list.

replace public interface RecordConverter with TimestampedBytesStore
extract value oldFormat -> newFormat conversion into static method
adds a new internal RecordConverter that uses new static method

mjsax · 2019-01-26T10:01:48Z

Call for review @guozhangwang @bbejeck @vvcephei @ableegoldman

vvcephei

Hi @mjsax ; This looks very clean. Thank you!

bbejeck

Thanks for the update @mjsax LGTM.

guozhangwang

Left a single comment about the code hierarchy, otherwise lgtm.

guozhangwang · 2019-01-28T04:30:34Z

            store instanceof WrappedStateStore ? ((WrappedStateStore) store).inner() : store;
        final RecordConverter recordConverter =
-            stateStore instanceof RecordConverter ? (RecordConverter) stateStore : new DefaultRecordConverter();
+            stateStore instanceof TimestampedBytesStore ? RecordConverter.converter() : record -> record;


Hmm.. I thought the restorer / state manager would now be agnostic to the convert at all, since the inner store impl is responsible for extending the interface as well as calling its TimestampedBytesStore function internally. Why do we still need an internal RecordConverter class?

The new static method that is used by the stores, will translate on-disk data from old to new format.

We still need a RecordConverter that translates data from the changelog topic to the new format, because on restore, we put plain <byte[],byte[]> key-value pairs into the store, and the store expects those record to be in the new format.

This make we realize, the the implementation of RecordConverter is actually not correct in this PR -- it should not use the new static method (that inserts a -1 as timestamp), but it need to put the actual record timestamp... Will update the PR accordingly.

But can't we make this logic inside the prepareBatch call inside the internal store impl (of course we would require customized users to do so as well) so that the callers do not need to be aware of that?

No, because prepareBatch (or actually restoreAllInternal) takes KeyValue<byte[],byte[] but not a ConsumerRecord. Thus there is not timestamp information we can add to the value.

👍 The convertToTimestampedFormat function is purely for converting the already-stored data into the new binary format. It can't actually add any timestamp information, because we don't have it when we retrieve the old-format data. So the function is like (value) -> [-1,value].

In contrast, the state restorer needs to be able to insert the timestamp, so the function is like (value, timestamp) -> [timestamp, value].

Actually, looking at this code again, I see Matthias was right, the RecordConverter erroneously calls through to the convertToTimestampedFormat function. Oops! Good catch!

Ah, I just refreshed, and looked at the new code. It LGTM. Thanks!

@mjsax Thanks for the explanation! LGTM.

mjsax · 2019-01-29T17:26:02Z

Retest this please.

mjsax · 2019-01-29T20:31:41Z

java.lang.ArrayIndexOutOfBoundsException: -1
	at kafka.integration.KafkaServerTestHarness.killBroker(KafkaServerTestHarness.scala:146)
	at kafka.api.ConsumerBounceTest.checkCloseWithCoordinatorFailure(ConsumerBounceTest.scala:238)
	at kafka.api.ConsumerBounceTest.testClose(ConsumerBounceTest.scala:211)

Retest this please.

mjsax · 2019-01-30T08:48:12Z

java.lang.AssertionError: Too many quotaLimit calls Map(PRODUCE -> 1, FETCH -> 1, REQUEST -> 3)
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.assertTrue(Assert.java:41)
	at kafka.api.CustomQuotaCallbackTest.testCustomQuotaCallback(CustomQuotaCallbackTest.scala:105)

java.lang.AssertionError: Client with id=QuotasTestProducer-1 should have been throttled
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.assertTrue(Assert.java:41)
	at kafka.api.QuotaTestClients.verifyThrottleTimeMetric(BaseQuotaTest.scala:229)
	at kafka.api.QuotaTestClients.verifyProduceThrottle(BaseQuotaTest.scala:215)
	at kafka.api.BaseQuotaTest.testThrottledProducerConsumer(BaseQuotaTest.scala:82)

java.lang.AssertionError: Client with id=QuotasTestProducer-1 should have been throttled
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.assertTrue(Assert.java:41)
	at kafka.api.QuotaTestClients.verifyThrottleTimeMetric(BaseQuotaTest.scala:229)
	at kafka.api.QuotaTestClients.verifyProduceThrottle(BaseQuotaTest.scala:215)
	at kafka.api.BaseQuotaTest.testQuotaOverrideDelete(BaseQuotaTest.scala:124)

\cc @ijuma Build is quite unstable. Can we look into this: https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/2048/

Retest this please.

mjsax · 2019-01-30T17:35:50Z

java.lang.AssertionError: Condition not met within timeout 15000. Streams never started.
	at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:365)
	at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:325)
	at org.apache.kafka.streams.KafkaStreamsTest.testStateThreadClose(KafkaStreamsTest.java:254)

and

java.lang.AssertionError: Condition not met within timeout 15000. Streams never started.
	at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:365)
	at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:325)
	at org.apache.kafka.streams.KafkaStreamsTest.shouldThrowOnCleanupWhileRunning(KafkaStreamsTest.java:549)

@guozhangwang Java11 passed. Run before Java8 passed. I tried retesting multiple times but test a flaky. Do you think it's ok to merge?

guozhangwang · 2019-01-30T23:53:21Z

@guozhangwang Java11 passed. Run before Java8 passed. I tried retesting multiple times but test a flaky. Do you think it's ok to merge?

LGTM.

* AK/trunk: fix typo (apache#5150) MINOR: Reduce replica.fetch.backoff.ms in ReassignPartitionsClusterTest (apache#5887) KAFKA-7766: Fail fast PR builds (apache#6059) KAFKA-7798: Expose embedded clientIds (apache#6107) KAFKA-7641; Introduce "group.max.size" config to limit group sizes (apache#6163) KAFKA-7433; Introduce broker options in TopicCommand to use AdminClient (KIP-377) MINOR: Fix some field definitions for ListOffsetReponse (apache#6214) KAFKA-7873; Always seek to beginning in KafkaBasedLog (apache#6203) KAFKA-7719: Improve fairness in SocketServer processors (KIP-402) (apache#6022) MINOR: fix checkstyle suppressions for generated RPC code to work on Windows KAFKA-7859: Use automatic RPC generation in LeaveGroups (apache#6188) KAFKA-7652: Part II; Add single-point query for SessionStore and use for flushing / getter (apache#6161) KAFKA-3522: Add RocksDBTimestampedStore (apache#6149) KAFKA-3522: Replace RecordConverter with TimestampedBytesStore (apache#6204)

…e#6204) Reviewers: John Roesler <john@confluent.io>, Bill Bejeck <bill@confluent.io>, Guozhang Wang <guozhang@confluent.io>

KAFKA-3522: Replace RecrodConverter with TimestampedBytesStore

f6b402c

mjsax added the streams label Jan 26, 2019

vvcephei approved these changes Jan 27, 2019

View reviewed changes

bbejeck approved these changes Jan 28, 2019

View reviewed changes

guozhangwang mentioned this pull request Jan 28, 2019

KAFKA-3522: Add RocksDBTimestampedStore #6149

Merged

guozhangwang reviewed Jan 28, 2019

View reviewed changes

Github comments

a2f34bb

guozhangwang approved these changes Jan 29, 2019

View reviewed changes

mjsax mentioned this pull request Jan 29, 2019

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

Merged

mjsax merged commit 201022d into apache:trunk Jan 30, 2019

mjsax deleted the kafka-3522-rocksdb-format-add-timestampedbytestore branch January 30, 2019 23:57

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

Conversation

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

Uh oh!

Uh oh!

mjsax commented Jan 26, 2019

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

bbejeck left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang Jan 28, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 29, 2019

Choose a reason for hiding this comment

Uh oh!

guozhangwang Jan 29, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax Jan 29, 2019

Choose a reason for hiding this comment

Uh oh!

vvcephei Jan 29, 2019

Choose a reason for hiding this comment

Uh oh!

vvcephei Jan 29, 2019

Choose a reason for hiding this comment

Uh oh!

guozhangwang Jan 29, 2019

Choose a reason for hiding this comment

Uh oh!

mjsax commented Jan 29, 2019

Uh oh!

mjsax commented Jan 29, 2019

Uh oh!

mjsax commented Jan 30, 2019

Uh oh!

mjsax commented Jan 30, 2019

Uh oh!

guozhangwang commented Jan 30, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

mjsax commented Jan 26, 2019 •

edited

Loading