You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/03/22 05:16:14 UTC
[GitHub] [kafka] vvcephei opened a new pull request #11926: KAFKA-13714: Fix cache flush position
vvcephei opened a new pull request #11926:
URL: https://github.com/apache/kafka/pull/11926
The caching store layers were passing down writes into lower store layers upon eviction, but not setting the context to the evicted records' context. Instead, the context was from whatever unrelated record was being processed at the time.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #11926: KAFKA-13714: Fix cache flush position
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11926:
URL: https://github.com/apache/kafka/pull/11926#issuecomment-1077021766
Test failures unrelated:
```
Build / JDK 8 and Scala 2.12 / org.apache.kafka.common.network.SslTransportLayerTest.[1] tlsProtocol=TLSv1.2, useInlinePem=false | 0.38 sec | 1
-- | -- | --
Build / JDK 8 and Scala 2.12 / integration.kafka.server.FetchRequestBetweenDifferentIbpTest.testControllerNewIBP()
[Build / JDK 8 and Scala 2.12 / org.apache.kafka.common.network.SslTransportLayerTest.[1] tlsProtocol=TLSv1.2, useInlinePem=false](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-11926/5/testReport/org.apache.kafka.common.network/SslTransportLayerTest/Build___JDK_8_and_Scala_2_12____1__tlsProtocol_TLSv1_2__useInlinePem_false/) 0.38 sec [1](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-11926/5/)
[Build / JDK 8 and Scala 2.12 / integration.kafka.server.FetchRequestBetweenDifferentIbpTest.testControllerNewIBP()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-11926/5/testReport/integration.kafka.server/FetchRequestBetweenDifferentIbpTest/Build___JDK_8_and_Scala_2_12___testControllerNewIBP__/)
```
The java 17 build is still queued after 5.5 hours.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on pull request #11926: KAFKA-13714: Fix cache flush position
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #11926:
URL: https://github.com/apache/kafka/pull/11926#issuecomment-1077034307
Cherry-picked to 3.2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei merged pull request #11926: KAFKA-13714: Fix cache flush position
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #11926:
URL: https://github.com/apache/kafka/pull/11926
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #11926: KAFKA-13714: Fix cache flush position
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11926:
URL: https://github.com/apache/kafka/pull/11926#discussion_r833701992
##########
File path: build.gradle
##########
@@ -207,7 +207,7 @@ if (file('.git').exists()) {
} else {
rat.enabled = false
}
-println("Starting build with version $version (commit id ${commitId.take(8)}) using Gradle $gradleVersion, Java ${JavaVersion.current()} and Scala ${versions.scala}")
+println("Starting build with version $version (commit id ${commitId == null ? "null" : commitId.take(8)}) using Gradle $gradleVersion, Java ${JavaVersion.current()} and Scala ${versions.scala}")
Review comment:
For some reason, I was getting an NPE here while running the tests locally (once). I could revert this change, but it also seems harmless to keep it.
##########
File path: build.gradle
##########
@@ -468,6 +474,12 @@ subprojects {
maxRetries = userMaxTestRetries
maxFailures = userMaxTestRetryFailures
}
+
+ // Allows devs to run tests in a loop to debug flaky tests
+ // Eg: I=0; while ./gradlew :streams:test -Prerun-tests --tests org.apache.kafka.streams.integration.IQv2StoreIntegrationTest --fail-fast; do (( I=$I+1 )); echo
+ if (project.hasProperty("rerun-tests")) {
+ outputs.upToDateWhen { false }
+ }
Review comment:
Same thing for integration tests.
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -226,10 +226,9 @@ private void putAndMaybeForward(final ThreadCache.DirtyEntry entry,
if (rawNewValue != null || rawOldValue != null) {
// we need to get the old values if needed, and then put to store, and then flush
final ProcessorRecordContext current = context.recordContext();
- context.setRecordContext(entry.entry().context());
- wrapped().put(entry.key(), entry.newValue());
-
try {
+ context.setRecordContext(entry.entry().context());
+ wrapped().put(entry.key(), entry.newValue());
Review comment:
we should be doing this inside the try block so that we will for sure set the context back no matter what happens.
##########
File path: build.gradle
##########
@@ -435,6 +435,12 @@ subprojects {
maxRetries = userMaxTestRetries
maxFailures = userMaxTestRetryFailures
}
+
+ // Allows devs to run tests in a loop to debug flaky tests
+ // Eg: I=0; while ./gradlew :streams:test -Prerun-tests --tests org.apache.kafka.streams.integration.IQv2StoreIntegrationTest --fail-fast; do (( I=$I+1 )); echo
+ if (project.hasProperty("rerun-tests")) {
+ outputs.upToDateWhen { false }
+ }
Review comment:
This is how I was able to repro the issue consistently, even though it only shows up roughly 1/500th of the time. I'll update the readme also to mention this approach instead of cleanTest, which causes us to recompile the tests on every iteration.
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
##########
@@ -142,7 +142,9 @@ public static void updatePosition(
if (stateStoreContext != null && stateStoreContext.recordMetadata().isPresent()) {
final RecordMetadata meta = stateStoreContext.recordMetadata().get();
- position.withComponent(meta.topic(), meta.partition(), meta.offset());
+ if (meta.topic() != null) {
Review comment:
Fixes some NPEs from the unit tests. I figure it's fine to have this here, since there's technically no contract that says topic can't be null and we're already guarding against a null context and missing recordMetadata.
##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java
##########
@@ -760,45 +760,50 @@ public static void after() {
@Test
public void verifyStore() {
- if (storeToTest.global()) {
- // See KAFKA-13523
- globalShouldRejectAllQueries();
- } else {
- shouldRejectUnknownQuery();
- shouldCollectExecutionInfo();
- shouldCollectExecutionInfoUnderFailure();
-
- if (storeToTest.keyValue()) {
- if (storeToTest.timestamped()) {
- final Function<ValueAndTimestamp<Integer>, Integer> valueExtractor =
- ValueAndTimestamp::value;
- shouldHandleKeyQuery(2, valueExtractor, 2);
- shouldHandleRangeQueries(valueExtractor);
- } else {
- final Function<Integer, Integer> valueExtractor = Function.identity();
- shouldHandleKeyQuery(2, valueExtractor, 2);
- shouldHandleRangeQueries(valueExtractor);
+ try {
+ if (storeToTest.global()) {
+ // See KAFKA-13523
+ globalShouldRejectAllQueries();
+ } else {
+ shouldRejectUnknownQuery();
+ shouldCollectExecutionInfo();
+ shouldCollectExecutionInfoUnderFailure();
+
+ if (storeToTest.keyValue()) {
+ if (storeToTest.timestamped()) {
+ final Function<ValueAndTimestamp<Integer>, Integer> valueExtractor =
+ ValueAndTimestamp::value;
+ shouldHandleKeyQuery(2, valueExtractor, 2);
+ shouldHandleRangeQueries(valueExtractor);
+ } else {
+ final Function<Integer, Integer> valueExtractor = Function.identity();
+ shouldHandleKeyQuery(2, valueExtractor, 2);
+ shouldHandleRangeQueries(valueExtractor);
+ }
}
- }
- if (storeToTest.isWindowed()) {
- if (storeToTest.timestamped()) {
- final Function<ValueAndTimestamp<Integer>, Integer> valueExtractor =
+ if (storeToTest.isWindowed()) {
+ if (storeToTest.timestamped()) {
+ final Function<ValueAndTimestamp<Integer>, Integer> valueExtractor =
ValueAndTimestamp::value;
- shouldHandleWindowKeyQueries(valueExtractor);
- shouldHandleWindowRangeQueries(valueExtractor);
- } else {
- final Function<Integer, Integer> valueExtractor = Function.identity();
- shouldHandleWindowKeyQueries(valueExtractor);
- shouldHandleWindowRangeQueries(valueExtractor);
+ shouldHandleWindowKeyQueries(valueExtractor);
+ shouldHandleWindowRangeQueries(valueExtractor);
+ } else {
+ final Function<Integer, Integer> valueExtractor = Function.identity();
+ shouldHandleWindowKeyQueries(valueExtractor);
+ shouldHandleWindowRangeQueries(valueExtractor);
+ }
}
- }
- if (storeToTest.isSession()) {
- // Note there's no "timestamped" differentiation here.
- // Idiosyncratically, SessionStores are _never_ timestamped.
- shouldHandleSessionKeyQueries();
+ if (storeToTest.isSession()) {
+ // Note there's no "timestamped" differentiation here.
+ // Idiosyncratically, SessionStores are _never_ timestamped.
+ shouldHandleSessionKeyQueries();
+ }
}
+ } catch (final AssertionError e) {
+ LOG.error("Failed assertion", e);
+ throw e;
Review comment:
Wrapping this and adding a log message helped me to correlate the actual moment of the test evaluation with the rest of the app logs in the test output. I figured I'd just leave it in and maybe help future generations.
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
##########
@@ -241,7 +240,13 @@ private void putAndMaybeForward(final ThreadCache.DirtyEntry entry,
}
}
} else {
- wrapped().put(entry.key(), entry.newValue());
Review comment:
This is why there was a bug under some configurations. We're leaving the wrong configuration in place while calling put on the wrapped store. We should set it to the evicted record's context before calling put. The same change is applied to the other two caching stores in this PR.
##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
##########
@@ -326,8 +326,8 @@ public synchronized void put(final Bytes key,
public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {
try (final WriteBatch batch = new WriteBatch()) {
dbAccessor.prepareBatch(entries, batch);
- StoreQueryUtils.updatePosition(position, context);
write(batch);
+ StoreQueryUtils.updatePosition(position, context);
Review comment:
Minor fix. In all other places, we update the position after the write, not before.
##########
File path: README.md
##########
@@ -37,13 +37,16 @@ Follow instructions in https://kafka.apache.org/quickstart
./gradlew integrationTest
### Force re-running tests without code change ###
- ./gradlew cleanTest test
- ./gradlew cleanTest unitTest
- ./gradlew cleanTest integrationTest
+ ./gradlew -Prerun-tests test
+ ./gradlew -Prerun-tests unitTest
+ ./gradlew -Prerun-tests integrationTest
### Running a particular unit/integration test ###
./gradlew clients:test --tests RequestResponseTest
+### Repeatedly running a particular unit/integration test ###
+ I=0; while ./gradlew clients:test -Prerun-tests --tests RequestResponseTest --fail-fast; do (( I=$I+1 )); echo "Completed run: $I"; sleep 1; done
Review comment:
This is how I was able to reproduce the issue in only a few hours, even though it failed only about 1/500th of the time.
##########
File path: streams/src/test/java/org/apache/kafka/streams/state/internals/CachingInMemoryKeyValueStoreTest.java
##########
@@ -221,17 +223,44 @@ public void shouldPutGetToFromCache() {
}
@Test
- public void shouldMatchPositionAfterPut() {
+ public void shouldMatchPositionAfterPutWithFlushListener() {
+ store.setFlushListener(record -> {}, false);
+ shouldMatchPositionAfterPut();
+ }
+
+ @Test
+ public void shouldMatchPositionAfterPutWithoutFlushListener() {
+ store.setFlushListener(null, false);
+ shouldMatchPositionAfterPut();
+ }
Review comment:
If you look at the caching stores above, you'll see they have separate branches, depending on whether there's a flush listener defined, so we need to test both branches. Ditto with the other tests.
##########
File path: streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java
##########
@@ -1350,7 +1355,7 @@ private static Properties streamsConfiguration(final boolean cache, final boolea
final String supplier, final String kind) {
final String safeTestName =
IQv2StoreIntegrationTest.class.getName() + "-" + cache + "-" + log + "-" + supplier
- + "-" + kind;
+ + "-" + kind + "-" + RANDOM.nextInt();
Review comment:
Allows us to iterate quickly on the tests without colliding on the state stores or waiting for the session timeout.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org