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/06/13 22:16:06 UTC
[GitHub] [kafka] mdedetrich opened a new pull request, #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
mdedetrich opened a new pull request, #12289:
URL: https://github.com/apache/kafka/pull/12289
Previously the tests would fail even on retryable errors. This change adds a retryUntil similar to the current until method however it will retry the test if it throws an exception that passes a predicate. The exception predicate has also been updated to handle another case that can occur due to a stream not being initialized.
In summary there are 2 reasons why this test is flaky. The first reason is that the test uses the `until` method which just fails whenever a retryable error occurs. Instead we should ideally retry the test only if the exception that is thrown matches a predicate defining a retryable error with this predicate being defined in the current `retrievableException`. The new `retryUntil` function respects the same deadlines/timeouts as the current `until` method.
The second issue is that the list of current predicates, i.e.
```java
containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
containsString("The state store, source-table, may have migrated to another instance"),
containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING")
```
wasn't complete. Another one needed to be added, namely "The specified partition 1 for store source-table does not exist.". This is because its possible for
```java
assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
```
or
```java
assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
```
(depending on which branch) to be thrown, i.e. see
```
org.apache.kafka.streams.errors.InvalidStateStorePartitionException: The specified partition 1 for store source-table does not exist.
at org.apache.kafka.streams.state.internals.WrappingStoreProvider.stores(WrappingStoreProvider.java:63)
at org.apache.kafka.streams.state.internals.CompositeReadOnlyKeyValueStore.get(CompositeReadOnlyKeyValueStore.java:53)
at org.apache.kafka.streams.integration.StoreQueryIntegrationTest.lambda$shouldQuerySpecificActivePartitionStores$5(StoreQueryIntegrationTest.java:223)
at org.apache.kafka.streams.integration.StoreQueryIntegrationTest.retryUntil(StoreQueryIntegrationTest.java:579)
at org.apache.kafka.streams.integration.StoreQueryIntegrationTest.shouldQuerySpecificActivePartitionStores(StoreQueryIntegrationTest.java:186)
```
This happens when the stream hasn't been initialized yet. I have run the test around 12k times using Intellij's JUnit testing framework without any exceptions
It seems reasonable that the other tests in `StoreQueryIntegrationTest.java` should also be using the new `retryUntil` rather than `until` if they also experience the same flaky tests however if this is the case then it should be done in future individual PR's.
### 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] mdedetrich commented on pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on PR #12289:
URL: https://github.com/apache/kafka/pull/12289#issuecomment-1173429678
@cadonna
Indeed you are correct, the real cause of the flakiness was not having the extra `"The specified partition 1 for store source-table does not exist."` condition. I reverted back to using `until()` and I got 10k runs without any flaky errors.
The latest commit has `retryUntil` removed and the test is using `until` just like before. I did however keep some of the refactoring that I did along with your suggested formatting/naming changes, let me know if you also want to revert some f these changes.
--
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] cadonna commented on pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12289:
URL: https://github.com/apache/kafka/pull/12289#issuecomment-1174059206
The timeout in the ARM build seems to be unrelated since that build does also time out a lot on other PRs recently.
--
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] cadonna commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r911825201
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -180,7 +183,7 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
// Assert that all messages in the first batch were processed in a timely manner
assertThat(semaphore.tryAcquire(batch1NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));
- until(() -> {
+ retryUntil(() -> {
Review Comment:
In the PR description you state:
> The first reason is that the test uses the until method which just fails whenever a retryable error occurs.
I doubt this statement, because within the `until()` the code verifies for retrieable exceptions and returns `false` if it encounters a retrieable exception which would trigger a further iteration over the code passed into the `until()`.
You can verify this behavior by running the following test:
```
public void testUntil() throws Exception {
until(() -> {
try {
throw new InvalidStateStorePartitionException(
"The specified partition 1 for store source-table does not exist.");
} catch (final InvalidStateStoreException exception) {
verifyRetriableException(exception);
LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
return false;
}
});
}
```
This code will retry throwing the exception until the timeout is exceeded.
You will notice that when the timeout is exceeded, the test passes. This is definitely a bug!
The following rewrite to `until()` will fix this bug:
```
private static void until(final TestCondition condition) {
boolean success = false;
final long deadline = System.currentTimeMillis() + IntegrationTestUtils.DEFAULT_TIMEOUT;
boolean deadlineExceeded = System.currentTimeMillis() >= deadline;
while (!success && !deadlineExceeded) {
try {
success = condition.conditionMet();
Thread.sleep(500L);
} catch (final RuntimeException e) {
throw e;
} catch (final Exception e) {
throw new RuntimeException(e);
} finally {
deadlineExceeded = System.currentTimeMillis() >= deadline;
}
}
if (deadlineExceeded) {
fail("Test execution timed out");
}
}
```
In summary, I think method `until()` does not cause any flakiness.
If you agree with my argumentation could you remove again `retryUntil()`, add the fix to `until()`, and adapt the PR description?
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
+ assertThat(store1.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
- assertThat(
+
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- } else {
- assertThat(store2.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ );
+ } else {
+ assertThat(store2.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
- assertThat(
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- }
- return true;
- } catch (final InvalidStateStoreException exception) {
- verifyRetrievableException(exception);
- LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
- return false;
+ );
}
- });
+ return true;
Review Comment:
See my comment above about `until()`.
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -535,17 +532,26 @@ public void shouldQueryStoresAfterAddingAndRemovingStreamThread() throws Excepti
});
}
+ private Matcher<String> retrievableException() {
+ return is(
+ anyOf(
+ containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
+ containsString("The state store, source-table, may have migrated to another instance"),
+ containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING"),
+ containsString("The specified partition 1 for store source-table does not exist.")
Review Comment:
I looked again at this code and discussed it with others and I think that might indeed be one cause of the flakiness of the test. My previous concerns proofed wrong. Thank you a lot for the investigation.
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
Review Comment:
We cannot merge the first two if-statements because in the first if-statement `store1` and `store2` are assigned and both variables are used in each of the branches in the second if-statement.
AFAIS, we could merge the second and the third if-statement. I leave it up to you whether the two if-statements should be merged or not. I am fine either way, with a slight preference to merge them. But I also understand that this is not really needed in this PR.
--
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] mdedetrich commented on pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on PR #12289:
URL: https://github.com/apache/kafka/pull/12289#issuecomment-1173691621
@cadonna Apologies, missed the part regarding `until`, there was also some issues regarding rebasing. I have fixed the `until` method as you stated as well as the various formatting issues and I checked that locally it passes checkstyle.
PR description is also updated, there is no mention of `retryUntil`. Not sure if you also want me to update the description of the original commit?
--
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] cadonna commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r912802169
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -221,24 +224,24 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
assertThat(store1.get(key), is(notNullValue()));
assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
final InvalidStateStoreException exception =
- assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
+ assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
assertThat(
- exception.getMessage(),
- containsString("The specified partition 1 for store source-table does not exist.")
+ exception.getMessage(),
+ containsString("The specified partition 1 for store source-table does not exist.")
Review Comment:
nit:
```suggestion
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,8 +213,8 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
Review Comment:
nit: we usually indent 4 spaces.
```suggestion
StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
.withPartition(keyDontBelongPartition);
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -221,24 +224,24 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
assertThat(store1.get(key), is(notNullValue()));
assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
final InvalidStateStoreException exception =
- assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
+ assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
assertThat(
- exception.getMessage(),
- containsString("The specified partition 1 for store source-table does not exist.")
+ exception.getMessage(),
+ containsString("The specified partition 1 for store source-table does not exist.")
);
} else {
assertThat(store2.get(key), is(notNullValue()));
assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
final InvalidStateStoreException exception =
- assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
+ assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
assertThat(
- exception.getMessage(),
- containsString("The specified partition 1 for store source-table does not exist.")
+ exception.getMessage(),
+ containsString("The specified partition 1 for store source-table does not exist.")
Review Comment:
nit:
```suggestion
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -221,24 +224,24 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
assertThat(store1.get(key), is(notNullValue()));
assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
final InvalidStateStoreException exception =
- assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
+ assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
Review Comment:
nit
```suggestion
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -221,24 +224,24 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
assertThat(store1.get(key), is(notNullValue()));
assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
final InvalidStateStoreException exception =
- assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
+ assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
assertThat(
- exception.getMessage(),
- containsString("The specified partition 1 for store source-table does not exist.")
+ exception.getMessage(),
+ containsString("The specified partition 1 for store source-table does not exist.")
);
} else {
assertThat(store2.get(key), is(notNullValue()));
assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
final InvalidStateStoreException exception =
- assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
+ assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
Review Comment:
nit:
```suggestion
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
```
--
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] mdedetrich commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r906835871
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
+ assertThat(store1.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
- assertThat(
+
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- } else {
- assertThat(store2.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ );
+ } else {
+ assertThat(store2.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
- assertThat(
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- }
- return true;
- } catch (final InvalidStateStoreException exception) {
- verifyRetrievableException(exception);
- LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
- return false;
+ );
}
- });
+ return true;
Review Comment:
So the original reason for the return value is so that you can easily replace it with `until`. Regarding extending the `retryOnExceptionWithTimeout()`, the reason why `retryUntil` is handcoded is because its keeping the same unwrapping `RuntimeException` that the original `until` did (see https://github.com/apache/kafka/blob/4c9eeef5b2dff9a4f0977fbc5ac7eaaf930d0d0e/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java#L559-L563).
I don't know the original reason why it was done this way but the intention was to minimize the amount of changes as to not increase the flakiness (but this is a hunch). Are you happy with just removing the `return` value for now or do you also want to extend ``retryOnExceptionWithTimeout()`?
--
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] cadonna merged pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
cadonna merged PR #12289:
URL: https://github.com/apache/kafka/pull/12289
--
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] mdedetrich commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r906835871
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
+ assertThat(store1.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
- assertThat(
+
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- } else {
- assertThat(store2.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ );
+ } else {
+ assertThat(store2.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
- assertThat(
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- }
- return true;
- } catch (final InvalidStateStoreException exception) {
- verifyRetrievableException(exception);
- LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
- return false;
+ );
}
- });
+ return true;
Review Comment:
So the original reason for the return value is so that you can easily replace it with `until`. Regarding extending the `retryOnExceptionWithTimeout(), the reason why `retryUntil` is handcoded is because its keeping the same unwrapping `RuntimeException` that the original `until` did (see https://github.com/apache/kafka/blob/4c9eeef5b2dff9a4f0977fbc5ac7eaaf930d0d0e/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java#L559-L563).
I don't know the original reason why it was done this way but the intention was to minimize the amount of changes as to not increase the flakiness (but this is a hunch). Are you happy with just removing the `return` value for now or do you also want to extend ``retryOnExceptionWithTimeout()`?
--
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] mdedetrich commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r906835167
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
Review Comment:
We could, I have a suspicion that its done this way so the test is easier to read since the flow is more linear but I am not against merging the statements if you think its better.
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -535,17 +532,26 @@ public void shouldQueryStoresAfterAddingAndRemovingStreamThread() throws Excepti
});
}
+ private Matcher<String> retrievableException() {
+ return is(
+ anyOf(
+ containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
+ containsString("The state store, source-table, may have migrated to another instance"),
+ containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING"),
+ containsString("The specified partition 1 for store source-table does not exist.")
Review Comment:
Agreed but if this is the case (and its a consistent bug) it would hit the timeout. Tbh investigating this was a bit out of my depth, maybe its wise to do a separate PR if we can find a better solution?
--
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] mdedetrich commented on pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on PR #12289:
URL: https://github.com/apache/kafka/pull/12289#issuecomment-1161447431
@cadonna
--
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] cadonna commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r906002582
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
+ assertThat(store1.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
- assertThat(
+
Review Comment:
```suggestion
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
+ assertThat(store1.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
- assertThat(
+
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
Review Comment:
```suggestion
assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
```
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
Review Comment:
I am wondering why we have three times the same if-statement `if (kafkaStreams1IsActive)`. Can we not merge the three if-statements together?
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -535,17 +532,26 @@ public void shouldQueryStoresAfterAddingAndRemovingStreamThread() throws Excepti
});
}
+ private Matcher<String> retrievableException() {
+ return is(
+ anyOf(
+ containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
+ containsString("The state store, source-table, may have migrated to another instance"),
+ containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING"),
+ containsString("The specified partition 1 for store source-table does not exist.")
+ )
+ );
+ }
Review Comment:
Fix for indentation and typo
```suggestion
private Matcher<String> retriableException() {
return is(
anyOf(
containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
containsString("The state store, source-table, may have migrated to another instance"),
containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING"),
containsString("The specified partition 1 for store source-table does not exist.")
)
);
}
```
Could you please fix the typo in the whole class. It should be `retriable`. The typo was mostly existing before this PR?
Could you please also fix the indentation in the whole class? We use 4 spaces.
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -210,39 +213,33 @@ public void shouldQuerySpecificActivePartitionStores() throws Exception {
}
final StoreQueryParameters<ReadOnlyKeyValueStore<Integer, Integer>> storeQueryParam2 =
- StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
- .withPartition(keyDontBelongPartition);
-
- try {
- // Assert that key is not served when wrong specific partition is requested
- // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
- // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
- if (kafkaStreams1IsActive) {
- assertThat(store1.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ StoreQueryParameters.<ReadOnlyKeyValueStore<Integer, Integer>>fromNameAndType(TABLE_NAME, keyValueStore())
+ .withPartition(keyDontBelongPartition);
+ // Assert that key is not served when wrong specific partition is requested
+ // If kafkaStreams1 is active for keyPartition, kafkaStreams2 would be active for keyDontBelongPartition
+ // So, in that case, store3 would be null and the store4 would not return the value for key as wrong partition was requested
+ if (kafkaStreams1IsActive) {
+ assertThat(store1.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams2, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams1, storeQueryParam2).get(key));
- assertThat(
+
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- } else {
- assertThat(store2.get(key), is(notNullValue()));
- assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
- final InvalidStateStoreException exception =
+ );
+ } else {
+ assertThat(store2.get(key), is(notNullValue()));
+ assertThat(getStore(kafkaStreams1, storeQueryParam2).get(key), is(nullValue()));
+ final InvalidStateStoreException exception =
assertThrows(InvalidStateStoreException.class, () -> getStore(kafkaStreams2, storeQueryParam2).get(key));
- assertThat(
+ assertThat(
exception.getMessage(),
containsString("The specified partition 1 for store source-table does not exist.")
- );
- }
- return true;
- } catch (final InvalidStateStoreException exception) {
- verifyRetrievableException(exception);
- LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
- return false;
+ );
}
- });
+ return true;
Review Comment:
Why do we still need to return something? With your change we would retry until we do not encounter any exception anymore.
Maybe it would be better to extend `retryOnExceptionWithTimeout()` (see https://github.com/apache/kafka/blob/e4b3a3cdeb295fdd4c4434ec1a7ee77b66553ae0/clients/src/test/java/org/apache/kafka/test/TestUtils.java#L361) to include the exception for which to retry.
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -535,17 +532,26 @@ public void shouldQueryStoresAfterAddingAndRemovingStreamThread() throws Excepti
});
}
+ private Matcher<String> retrievableException() {
+ return is(
+ anyOf(
+ containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
+ containsString("The state store, source-table, may have migrated to another instance"),
+ containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING"),
+ containsString("The specified partition 1 for store source-table does not exist.")
Review Comment:
I see that this might happen, but I am wondering if we can solve that differently. The reason is that a bug could also throw that exception and we would just retry. Due to exceeding the timeout we would probably understand that it is a bug, but it seems really hard to investigate.
--
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] mdedetrich commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r906835016
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -535,17 +532,26 @@ public void shouldQueryStoresAfterAddingAndRemovingStreamThread() throws Excepti
});
}
+ private Matcher<String> retrievableException() {
+ return is(
+ anyOf(
+ containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
+ containsString("The state store, source-table, may have migrated to another instance"),
+ containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING"),
+ containsString("The specified partition 1 for store source-table does not exist.")
Review Comment:
> The reason is that a bug could also throw that exception and we would just retry.
Agreed but if this is the case (and its a consistent bug) it would hit the timeout. Tbh investigating this was a bit out of my depth, maybe its wise to do a separate PR if we can find a better solution?
--
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] mdedetrich commented on a diff in pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
mdedetrich commented on code in PR #12289:
URL: https://github.com/apache/kafka/pull/12289#discussion_r906838047
##########
streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java:
##########
@@ -535,17 +532,26 @@ public void shouldQueryStoresAfterAddingAndRemovingStreamThread() throws Excepti
});
}
+ private Matcher<String> retrievableException() {
+ return is(
+ anyOf(
+ containsString("Cannot get state store source-table because the stream thread is PARTITIONS_ASSIGNED, not RUNNING"),
+ containsString("The state store, source-table, may have migrated to another instance"),
+ containsString("Cannot get state store source-table because the stream thread is STARTING, not RUNNING"),
+ containsString("The specified partition 1 for store source-table does not exist.")
+ )
+ );
+ }
Review Comment:
Done
--
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] cadonna commented on pull request #12289: KAFKA-13957: Fix flaky shouldQuerySpecificActivePartitionStores test
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12289:
URL: https://github.com/apache/kafka/pull/12289#issuecomment-1173577446
@mdedetrich The builds are red due to a checkstyle issue. Could you please fix the issue?
--
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