YeonCheolGit commented on a change in pull request #11848:
URL: https://github.com/apache/kafka/pull/11848#discussion_r820060984
##########
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;
+ }
Review comment:
```suggestion
// we're at the end of the input.
if (queryResult.getResult().value() == batch1NumMessages -
1) return;
```
--
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
