QueryableIntegrationTest.queryOnRebalance transient failure
Don't produce messages on a separate thread continuosly. Just produce one of each value and stop.
Close the producer once finished.
Author: Damian Guy
Reviewers: Guozhang Wang Ismael Juma
Closes from dguy/qs-test
@@ -329,19 +329,17 @@ public class QueryableStateIntegrationTest {
final int numThreads = STREAM_TWO_PARTITIONS;
final StreamRunnable[] streamRunnables = new StreamRunnable[numThreads];
final Thread[] streamThreads = new Thread[numThreads];
- final int numIterations = 500000;
- // create concurrent producer
- final ProducerRunnable producerRunnable = new ProducerRunnable(streamThree, inputValues, numIterations);
- final Thread producerThread = new Thread(producerRunnable);
+ final ProducerRunnable producerRunnable = new ProducerRunnable(streamThree, inputValues, 1);
+ producerRunnable.run();
- // create three stream threads
+
+ // create stream threads
for (int i = 0; i < numThreads; i++) {
streamRunnables[i] = new StreamRunnable(streamThree, outputTopicThree, i);
streamThreads[i] = new Thread(streamRunnables[i]);
streamThreads[i].start();
}
- producerThread.start();
try {
waitUntilAtLeastNumRecordProcessed(outputTopicThree, 1);
@@ -375,9 +373,6 @@ public class QueryableStateIntegrationTest {
streamThreads[i].join();
}
}
- producerRunnable.shutdown();
- producerThread.interrupt();
- producerThread.join();
}
}
@@ -913,9 +908,8 @@ public class QueryableStateIntegrationTest {
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
- final KafkaProducer<String, String>
- producer =
- new KafkaProducer<>(producerConfig, new StringSerializer(), new StringSerializer());
+ try (final KafkaProducer<String, String> producer =
+ new KafkaProducer<>(producerConfig, new StringSerializer(), new StringSerializer())) {
while (getCurrIteration() < numIterations && !shutdown) {
for (int i = 0; i < inputValues.size(); i++) {
@@ -925,6 +919,7 @@ public class QueryableStateIntegrationTest {
}
}
}
+ }
}
