KAFKA-9176: Retry on getting local stores from KafkaStreams by guozhangwang · Pull Request #8568 · apache/kafka

guozhangwang · 2020-04-28T04:58:55Z

This PR fixes and improves two major issues:

When calling KafkaStreams#store we can always get an InvalidStateStoreException, and even waiting for Streams state to become RUNNING is not sufficient (this is also how OptimizedKTableIntegrationTest failed). So I wrapped all the function with a Util wrapper that captures and retries on that exception.
While trouble-shooting this issue, I also realized a potential bug in test-util's produceKeyValuesSynchronously, which creates a new producer for each of the record to send in that batch --- i.e. if you are sending N records with a single call, within that call it will create N producers used to send one record each, which is very slow and costly.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

guozhangwang

ping @vvcephei @ableegoldman for reviews.

guozhangwang · 2020-04-28T04:59:22Z

-            time.sleep(1L);
+                                                            final boolean enableTransactions) {
+
+        try (final Producer<K, V> producer = new KafkaProducer<>(producerConfig)) {


This is fix 2).

guozhangwang · 2020-04-28T04:59:36Z

        return waitUntilFinalKeyValueRecordsReceived(consumerConfig, topic, expectedRecords, waitTime, false);
    }

-    public static <K, V> List<KeyValueTimestamp<K, V>> waitUntilFinalKeyValueTimestampRecordsReceived(final Properties consumerConfig,


Those functions are not used anywhere, ditto below.

thanks for the cleanup

guozhangwang · 2020-04-28T05:00:29Z


-        // Assert that the current value in store reflects all messages being processed
-        assertThat(newActiveStore.get(key), is(equalTo(totalNumMessages - 1)));
+        TestUtils.retryOnExceptionWithTimeout(100, 60 * 1000, () -> {


This is a minor fix, that we should retry this condition.

guozhangwang · 2020-04-28T05:01:04Z

        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
        config.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
+        config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
+        config.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);


This is a fix to the test itself: with caching the records are delayed sending to the sink topics.

ableegoldman · 2020-04-28T17:52:25Z

@@ -163,8 +159,10 @@ public void shouldApplyUpdatesToStandbyStore() throws Exception {
        // Assert that all messages in the second batch were processed in a timely manner
        assertThat(semaphore.tryAcquire(batch2NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));


OptimizedKTableIntegrationTest.shouldApplyUpdatesToStandbyStore still failed on one of the builds at this line :/
But, at least we got farther into the test before it failed so I'd say this is still an improvement 😄

ableegoldman

Nice! One question about why we sometimes do a null check (and sometimes don't) but otherwise LGTM. Kind of a bummer that OptimizedKTableIntegrationTest.shouldApplyUpdatesToStandbyStore still failed on one of the builds but feel free to merge this as a first step and investigate the new failure separately

ableegoldman · 2020-04-28T17:59:41Z

-            kafkaStreams.store(StoreQueryParameters.fromNameAndType(globalStore, QueryableStoreTypes.keyValueStore()));
+        final ReadOnlyKeyValueStore<Long, String> replicatedStore = IntegrationTestUtils
+            .getStore(globalStore, kafkaStreams, QueryableStoreTypes.keyValueStore());
+        assertNotNull(replicatedStore);


Why do we have to check for null now?

Since previously we would just throw the exception with the un-wrapped call, here asserting it is not null is equal to make sure that the store is indeed returned.

guozhangwang · 2020-04-28T20:50:08Z

I looked at the three failed tests:

OptimizedKTableIntegrationTest.shouldApplyUpdatesToStandbyStore is actually due to the issue that KAFKA-9823: Follow-up, check state for handling commit error response #8548 tried to fix. Waiting for @vvcephei to review 8548
EosIntegrationTest.shouldNotViolateEosIfOneTaskFailsWithState[exactly_once_beta] is being looked at by @mjsax as KAFKA-9831
GlobalKTableEOSIntegrationTest.shouldKStreamGlobalKTableLeftJoin[exactly_once_beta] is a new issue, I created KAFKA-9928 for this, and my gut feeling is that it has the same root cause as KAFKA-9831. (also cc @mjsax )

So I think this PR is good to be merged.

vvcephei

Just a few minor comments, which you can take or leave :)

Thanks for the awesome improvement!

vvcephei · 2020-04-28T21:05:30Z

-
+                                  final Set<KeyValue<Long, Long>> expectedStoreContent) throws InterruptedException {
+        final ReadOnlyKeyValueStore<Long, Long> store = IntegrationTestUtils
+            .getStore(300000L, storeName, streams, QueryableStoreTypes.keyValueStore());


Suggested change

.getStore(300000L, storeName, streams, QueryableStoreTypes.keyValueStore());

.getStore(300_000L, storeName, streams, QueryableStoreTypes.keyValueStore());

vvcephei · 2020-04-28T21:12:18Z

+                    if (store == null)
+                        return false;


not a huge deal, but technically, these should have brackets.

vvcephei · 2020-04-28T21:14:22Z

-                final Future<RecordMetadata> f = producer.send(
-                    new ProducerRecord<>(topic, null, timestamp, record.key, record.value, headers));
-                f.get();
+                producer.send(new ProducerRecord<>(topic, null, timestamp, record.key, record.value, headers));


I guess the flush at the end makes it synchronous anyway?

Previously we wait after sending each record, here we only wait once after sending all records, so it is more efficient.

Thanks. That's what I was asking for confirmation on. I realize now the structure of my sentence was ambiguous.

I agree that the method contract is that the batch should be synchronously produced, not that each record should be synchronously produced, so this change looks good to me.

vvcephei · 2020-04-28T21:14:37Z

        return waitUntilFinalKeyValueRecordsReceived(consumerConfig, topic, expectedRecords, waitTime, false);
    }

-    public static <K, V> List<KeyValueTimestamp<K, V>> waitUntilFinalKeyValueTimestampRecordsReceived(final Properties consumerConfig,


thanks for the cleanup

…t-store-with-exception

guozhangwang · 2020-04-28T22:04:39Z

Merged to trunk.

…/master` * apache-github/trunk: (45 commits) MINOR: Fix broken JMX link in docs by adding missing starting double quote (apache#8587) KAFKA-9652: Fix throttle metric in RequestChannel and request log due to KIP-219 (apache#8567) KAFKA-9922: Update demo instructions in examples README (apache#8559) KAFKA-9830: Implement AutoCloseable in ErrorReporter and subclasses (apache#8442) KAFKA-9875: Make integration tests more resilient (apache#8578) KAFKA-9932: Don't load configs from ZK when the log has already been loaded (apache#8582) KAFKA-9925: decorate pseudo-topics with app id (apache#8574) KAFKA-9832: fix attempt to commit non-running tasks (apache#8580) KAFKA-9127: don't create StreamThreads for global-only topology (apache#8540) MINOR: add support for kafka 2.4 and 2.5 to downgrade test KAFKA-9176: Retry on getting local stores from KafkaStreams (apache#8568) KAFKA-9823: Follow-up, check state for handling commit error response (apache#8548) KAFKA-6145: KIP-441: Add TaskAssignor class config (apache#8541) MINOR: Fix partition numbering from 0 to P-1 instead of P in docs (apache#8572) KAFKA-9921: disable caching on stores configured to retain duplicates (apache#8564) Minor: remove redundant check in auto preferred leader election (apache#8566) MINOR: Update the link to the Raft paper in docs (apache#8560) MINOR: Fix typos in config properties in MM2 test (apache#8561) MINOR: Improve producer test BufferPoolTest#testCloseNotifyWaiters. (apache#7982) MINOR: document how to escape json parameters to ducktape tests (apache#8546) ...

There was a minor conflict in gradle.properties because the default Scala version changed upstream to Scala 2.13. I kept the upstream change. Related to this, I have updated Jenkinsfile to compile and validate with Scala 2.12 in a separate stage so that we ensure we maintain compatibility. Unlike Apache Kafka, we only run the tests with the default Scala version, which is now 2.13. * apache-github/trunk: (45 commits) MINOR: Fix broken JMX link in docs by adding missing starting double quote (apache#8587) KAFKA-9652: Fix throttle metric in RequestChannel and request log due to KIP-219 (apache#8567) KAFKA-9922: Update demo instructions in examples README (apache#8559) KAFKA-9830: Implement AutoCloseable in ErrorReporter and subclasses (apache#8442) KAFKA-9875: Make integration tests more resilient (apache#8578) KAFKA-9932: Don't load configs from ZK when the log has already been loaded (apache#8582) KAFKA-9925: decorate pseudo-topics with app id (apache#8574) KAFKA-9832: fix attempt to commit non-running tasks (apache#8580) KAFKA-9127: don't create StreamThreads for global-only topology (apache#8540) MINOR: add support for kafka 2.4 and 2.5 to downgrade test KAFKA-9176: Retry on getting local stores from KafkaStreams (apache#8568) KAFKA-9823: Follow-up, check state for handling commit error response (apache#8548) KAFKA-6145: KIP-441: Add TaskAssignor class config (apache#8541) MINOR: Fix partition numbering from 0 to P-1 instead of P in docs (apache#8572) KAFKA-9921: disable caching on stores configured to retain duplicates (apache#8564) Minor: remove redundant check in auto preferred leader election (apache#8566) MINOR: Update the link to the Raft paper in docs (apache#8560) MINOR: Fix typos in config properties in MM2 test (apache#8561) MINOR: Improve producer test BufferPoolTest#testCloseNotifyWaiters. (apache#7982) MINOR: document how to escape json parameters to ducktape tests (apache#8546) ...

wrap around getting stores

f936848

guozhangwang commented Apr 28, 2020

View reviewed changes

ableegoldman reviewed Apr 28, 2020

View reviewed changes

ableegoldman approved these changes Apr 28, 2020

View reviewed changes

vvcephei approved these changes Apr 28, 2020

View reviewed changes

guozhangwang added 2 commits April 28, 2020 14:44

Merge branch 'trunk' of https://github.com/apache/kafka into K9176-ge…

4bb8463

…t-store-with-exception

github comments

57a2c39

guozhangwang merged commit 11fc953 into apache:trunk Apr 28, 2020

guozhangwang deleted the K9176-get-store-with-exception branch April 28, 2020 22:04

		@@ -163,8 +159,10 @@ public void shouldApplyUpdatesToStandbyStore() throws Exception {
		// Assert that all messages in the second batch were processed in a timely manner
		assertThat(semaphore.tryAcquire(batch2NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));

	.getStore(300000L, storeName, streams, QueryableStoreTypes.keyValueStore());
	.getStore(300_000L, storeName, streams, QueryableStoreTypes.keyValueStore());

Conversation

guozhangwang commented Apr 28, 2020

Committer Checklist (excluded from commit message)

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!

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!

ableegoldman 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!

guozhangwang commented Apr 28, 2020

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Apr 28, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants