vvcephei commented on a change in pull request #11848:
URL: https://github.com/apache/kafka/pull/11848#discussion_r822019343
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/ConsistencyVectorIntegrationTest.java
##########
@@ -131,67 +129,56 @@ public void shouldHaveSamePositionBoundActiveAndStandBy()
throws Exception {
final QueryableStoreType<ReadOnlyKeyValueStore<Integer, Integer>>
queryableStoreType = keyValueStore();
// Assert that both active and standby have the same position bound
- TestUtils.waitForCondition(() -> {
- final ReadOnlyKeyValueStore<Integer, Integer> store1 =
getStore(TABLE_NAME, kafkaStreams1, true, queryableStoreType);
- return store1.get(key) == batch1NumMessages - 1;
- }, "store1 cannot find results for key");
- TestUtils.waitForCondition(() -> {
- final ReadOnlyKeyValueStore<Integer, Integer> store2 =
getStore(TABLE_NAME, kafkaStreams2, true, queryableStoreType);
- return store2.get(key) == batch1NumMessages - 1;
- }, "store2 cannot find results for key");
-
- final AtomicInteger count = new AtomicInteger();
- for (final TestingRocksDBStore store : supplier.stores) {
- if (store.getDbDir() != null) {
- assertThat(store.getDbDir().toString().contains("/0_0/"),
is(true));
-
assertThat(store.getPosition().getPartitionPositions(INPUT_TOPIC_NAME),
notNullValue());
-
assertThat(store.getPosition().getPartitionPositions(INPUT_TOPIC_NAME),
hasEntry(0, 99L));
- count.incrementAndGet();
- }
- }
- assertThat(count.get(), is(2));
- }
-
- public class TestingRocksDBStore extends RocksDBStore {
- public TestingRocksDBStore(final String name, final String
metricsScope) {
- super(name, metricsScope);
- }
-
- public Position getPosition() {
- return position;
- }
-
- public File getDbDir() {
- return dbDir;
- }
+ final StateQueryRequest<ValueAndTimestamp<Integer>> request =
+ StateQueryRequest
+ .inStore(TABLE_NAME)
+ .withQuery(KeyQuery.<Integer,
ValueAndTimestamp<Integer>>withKey(key))
+ .withPositionBound(PositionBound.unbounded());
+
+ checkPosition(batch1NumMessages, request, kafkaStreams1);
+ checkPosition(batch1NumMessages, request, kafkaStreams2);
}
- public class TestingRocksDbKeyValueBytesStoreSupplier implements
KeyValueBytesStoreSupplier {
-
- public List<TestingRocksDBStore> stores = new LinkedList<>();
-
- private final String name;
-
- public TestingRocksDbKeyValueBytesStoreSupplier(final String name) {
- this.name = name;
- }
-
- @Override
- public String name() {
- return name;
- }
+ private void checkPosition(final int batch1NumMessages,
+ final
StateQueryRequest<ValueAndTimestamp<Integer>> request,
+ final KafkaStreams kafkaStreams1) throws
InterruptedException {
+ final long maxWaitMs = TestUtils.DEFAULT_MAX_WAIT_MS;
+ final long expectedEnd = System.currentTimeMillis() + maxWaitMs;
+
+ while (true) {
+ final StateQueryResult<ValueAndTimestamp<Integer>>
stateQueryResult =
+ IntegrationTestUtils.iqv2WaitForResult(
+ kafkaStreams1,
+ request
+ );
+ final QueryResult<ValueAndTimestamp<Integer>> queryResult =
+ stateQueryResult.getPartitionResults().get(0);
+ if (queryResult.isSuccess() && queryResult.getResult() != null) {
+ // invariant: each value is also at the equivalent offset
+ assertThat(
+ "Result:" + queryResult,
+ queryResult.getPosition(),
+ is(
+ Position.emptyPosition()
+ .withComponent(INPUT_TOPIC_NAME, 0,
queryResult.getResult().value())
+ )
+ );
+
+ if (queryResult.getResult().value() == batch1NumMessages - 1) {
+ // we're at the end of the input.
+ return;
+ }
+ } else {
+ if (expectedEnd <= System.currentTimeMillis()) {
+ throw new RuntimeException(
+ "Test timed out in " + maxWaitMs);
+ }
+ }
- @Override
- public KeyValueStore<Bytes, byte[]> get() {
- final TestingRocksDBStore testingRocksDBStore = new
TestingRocksDBStore(name, metricsScope());
- stores.add(testingRocksDBStore);
- return testingRocksDBStore;
+ // we're not done yet, so sleep a bit and test again.
+ Thread.sleep(maxWaitMs / 10);
Review comment:
Yep. This is more attempts than the previous utility, which I think
would only have done four retries, but it's the same overall time limit on the
test.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
