Skip to content

MINOR: Deflake Streams position restart integration test #23111

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ezhou413 wants to merge 1 commit into
base: trunk
Choose a base branch
from
+21 −15
Open

MINOR: Deflake Streams position restart integration test #23111

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 21 additions & 15 deletions ...ts/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,14 +110,20 @@ public class PositionRestartIntegrationTest {
private static final long SEED = new Random().nextLong();
private static final int NUM_BROKERS = 3;
public static final Duration WINDOW_SIZE = Duration.ofMinutes(5);
private static int port = 0;
private static final String INPUT_TOPIC_NAME = "input-topic";
private static final Position INPUT_POSITION = Position.emptyPosition();
private static final String STORE_NAME = "kv-store";
private static final long RECORD_TIME = System.currentTimeMillis();
private static final long WINDOW_START =
(RECORD_TIME / WINDOW_SIZE.toMillis()) * WINDOW_SIZE.toMillis();
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);

// The cluster and the position it produces must be per-instance, not static. Subclasses (e.g. the
// transactional variant) inherit this class's @BeforeAll/@AfterAll, so a static cluster would be
// shared across both test classes: whichever ran second would find an already-closed
// KafkaClusterTestKit, whose executor service cannot be restarted. Under PER_CLASS lifecycle each
// test class gets its own instance, and therefore its own cluster.
private final EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(NUM_BROKERS);
private final Position inputPosition = Position.emptyPosition();
private int port = 0;
private KafkaStreams kafkaStreams;

public enum StoresToTest {
Expand Down Expand Up @@ -281,16 +287,16 @@ protected Stream<Arguments> data() {
}

@BeforeAll
public static void before()
public void before()
throws InterruptedException, IOException, ExecutionException, TimeoutException {

CLUSTER.start();
CLUSTER.deleteAllTopics();
cluster.start();
cluster.deleteAllTopics();
final int partitions = 2;
CLUSTER.createTopic(INPUT_TOPIC_NAME, partitions, 1);
cluster.createTopic(INPUT_TOPIC_NAME, partitions, 1);

final Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);

Expand All @@ -315,15 +321,15 @@ public static void before()
for (final Future<RecordMetadata> future : futures) {
final RecordMetadata recordMetadata = future.get(1, TimeUnit.MINUTES);
assertThat(recordMetadata.hasOffset(), is(true));
INPUT_POSITION.withComponent(
inputPosition.withComponent(
recordMetadata.topic(),
recordMetadata.partition(),
recordMetadata.offset()
);
}
}

assertThat(INPUT_POSITION, equalTo(
assertThat(inputPosition, equalTo(
Position
.emptyPosition()
.withComponent(INPUT_TOPIC_NAME, 0, 1L)
Expand Down Expand Up @@ -366,8 +372,8 @@ public void afterTest() {
}

@AfterAll
public static void after() {
CLUSTER.stop();
public void after() {
cluster.stop();
}

@ParameterizedTest
Expand Down Expand Up @@ -405,12 +411,12 @@ private void shouldReachExpectedPosition(final Query<?> query) {
inStore(STORE_NAME)
.withQuery(query)
.withPartitions(Set.of(0, 1))
.withPositionBound(PositionBound.at(INPUT_POSITION));
.withPositionBound(PositionBound.at(inputPosition));

final StateQueryResult<?> result =
IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);

assertThat(result.getPosition(), is(INPUT_POSITION));
assertThat(result.getPosition(), is(inputPosition));
}

private static void setUpSessionDSLTopology(final SessionBytesStoreSupplier supplier,
Expand Down Expand Up @@ -668,7 +674,7 @@ protected Properties streamsConfiguration(final boolean cache,
config.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName);
config.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:" + (++port));
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
config.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
Expand Down