extract retry checker and update with retriable exception causing flaky StoreQueryIntegrationTest
Add a new case to the list of possible retriable exceptions for the flaky tests to take care of threads starting up
Reviewers: Leah Thomas Anna Sophie Blee-Goldman
@@ -148,16 +148,7 @@ public class StoreQueryIntegrationTest {
}
return true;
} catch (final InvalidStateStoreException exception) {
- assertThat(
- "Unexpected exception thrown while getting the value from store.",
- exception.getMessage(),
- 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")
- )
- )
- );
+ verifyRetrievableException(exception);
LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
return false;
}
@@ -241,16 +232,7 @@ public class StoreQueryIntegrationTest {
}
return true;
} catch (final InvalidStateStoreException exception) {
- assertThat(
- "Unexpected exception thrown while getting the value from store.",
- exception.getMessage(),
- 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")
- )
- )
- );
+ verifyRetrievableException(exception);
LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
return false;
}
@@ -513,16 +495,7 @@ public class StoreQueryIntegrationTest {
assertThat(store1.get(key3), is(notNullValue()));
return true;
} catch (final InvalidStateStoreException exception) {
- assertThat(
- "Unexpected exception thrown while getting the value from store.",
- exception.getMessage(),
- 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")
- )
- )
- );
+ verifyRetrievableException(exception);
LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
return false;
}
@@ -543,20 +516,25 @@ public class StoreQueryIntegrationTest {
assertThat(store1.get(key3), is(notNullValue()));
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;
+ }
+ });
+ }
+
+ private void verifyRetrievableException(final Exception exception) {
assertThat(
"Unexpected exception thrown while getting the value from store.",
exception.getMessage(),
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("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")
)
)
);
- LOG.info("Either streams wasn't running or a re-balancing took place. Will try again.");
- return false;
- }
- });
}
private static void until(final TestCondition condition) {
