Skip to content

KAFKA-7944: Improve Suppress test coverage #6382

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

Merged
guozhangwang merged 6 commits into from
Mar 12, 2019
Merged

KAFKA-7944: Improve Suppress test coverage #6382

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
63c4d7b
KAFKA-7944: Improve Suppress test coverage
vvcephei Mar 5, 2019
efb4407
remove unused imports
vvcephei Mar 6, 2019
a724897
fix durability test
vvcephei Mar 7, 2019
5ad02ea
split generator use cases and simplify code
vvcephei Mar 7, 2019
48dabcd
cr comments
vvcephei Mar 7, 2019
7420fd0
Remove IDEA warning suppression.
vvcephei Mar 8, 2019
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
7 changes: 5 additions & 2 deletions ...ms/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,14 @@

import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.tests.SmokeTestClient;
import org.apache.kafka.streams.tests.SmokeTestDriver;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Map;
import java.util.Properties;
Expand Down Expand Up @@ -53,7 +55,8 @@ private Driver(final String bootstrapServers, final int numKeys, final int maxRe
@Override
public void run() {
try {
final Map<String, Set<Integer>> allData = generate(bootstrapServers, numKeys, maxRecordsPerKey, true);
final Map<String, Set<Integer>> allData =
generate(bootstrapServers, numKeys, maxRecordsPerKey, Duration.ZERO, true);
result = verify(bootstrapServers, allData, maxRecordsPerKey);

} catch (final Exception ex) {
Expand All @@ -76,7 +79,7 @@ public void shouldWorkWithRebalance() throws InterruptedException {
int numClientsCreated = 0;
final ArrayList<SmokeTestClient> clients = new ArrayList<>();

CLUSTER.createTopics(SmokeTestDriver.topics());
IntegrationTestUtils.cleanStateBeforeTest(CLUSTER, SmokeTestDriver.topics());

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

More stable and future-proof test setup code. I needed this because I added a second test here, which I've since removed.

Leaving this change, though, so we won't have to waste time debugging again next time we try to add a test.


final String bootstrapServers = CLUSTER.bootstrapServers();
final Driver driver = new Driver(bootstrapServers, 10, 1000);
Expand Down
5 changes: 0 additions & 5 deletions .../test/java/org/apache/kafka/streams/integration/SuppressionDurabilityIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,11 +92,6 @@ public SuppressionDurabilityIntegrationTest(final boolean eosEnabled) {
this.eosEnabled = eosEnabled;
}

@Parameters(name = "{index}: eosEnabled={0}")
public static Collection<Object[]> parameters() {
return Arrays.asList(new Object[] {false}, new Object[] {true});
}

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

unused!

private KTable<String, Long> buildCountsTable(final String input, final StreamsBuilder builder) {
return builder
.table(
Expand Down
35 changes: 22 additions & 13 deletions streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,16 +32,19 @@
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.Suppressed;
import org.apache.kafka.streams.kstream.Suppressed.BufferConfig;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.test.TestUtils;

import java.time.Duration;
import java.time.Instant;
import java.util.Properties;

import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;

public class SmokeTestClient extends SmokeTestUtil {

private final String name;
Expand Down Expand Up @@ -113,7 +116,7 @@ private KafkaStreams createKafkaStreams(final Properties props) {
final Topology build = getTopology();
final KafkaStreams streamsClient = new KafkaStreams(build, getStreamsConfig(props));
streamsClient.setStateListener((newState, oldState) -> {
System.out.printf("%s: %s -> %s%n", name, oldState, newState);
System.out.printf("%s %s: %s -> %s%n", name, Instant.now(), oldState, newState);

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added this while debugging the smoke test. Printing the time allows us to correlate events across test nodes and with the test logs.

@guozhangwang guozhangwang Mar 7, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice find!

if (oldState == KafkaStreams.State.REBALANCING && newState == KafkaStreams.State.RUNNING) {
started = true;
}
Expand Down Expand Up @@ -149,24 +152,22 @@ public Topology getTopology() {
.withRetention(Duration.ofHours(25))
);

minAggregation
.toStream()
.filterNot((k, v) -> k.key().equals("flush"))
.map((key, value) -> new KeyValue<>(key.toString(), value))
.to("min-raw", Produced.with(stringSerde, intSerde));
streamify(minAggregation, "min-raw");

minAggregation
.suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
.toStream()
.filterNot((k, v) -> k.key().equals("flush"))
.map((key, value) -> new KeyValue<>(key.toString(), value))
.to("min-suppressed", Produced.with(stringSerde, intSerde));
streamify(minAggregation.suppress(untilWindowCloses(BufferConfig.unbounded())), "min-suppressed");

minAggregation
.toStream(new Unwindow<>())
.filterNot((k, v) -> k.equals("flush"))
.to("min", Produced.with(stringSerde, intSerde));

final KTable<Windowed<String>, Integer> smallWindowSum = groupedData
.windowedBy(TimeWindows.of(Duration.ofSeconds(2)).advanceBy(Duration.ofSeconds(1)).grace(Duration.ofSeconds(30)))
.reduce((l, r) -> l + r);

streamify(smallWindowSum, "sws-raw");
streamify(smallWindowSum.suppress(untilWindowCloses(BufferConfig.unbounded())), "sws-suppressed");

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a "small window sum" stream to the topology, which does a "normal" windowed computation. The exact data production timing is non-deterministic, so we can't verify the results, but we can verify that there is exactly one result per windowed key in the suppressed stream

final KTable<String, Integer> minTable = builder.table(
"min",
Consumed.with(stringSerde, intSerde),
Expand Down Expand Up @@ -250,4 +251,12 @@ public Topology getTopology() {

return builder.build();
}

private static void streamify(final KTable<Windowed<String>, Integer> windowedTable, final String topic) {
windowedTable
.toStream()
.filterNot((k, v) -> k.key().equals("flush"))
.map((key, value) -> new KeyValue<>(key.toString(), value))
.to(topic, Produced.with(stringSerde, intSerde));
}
}
62 changes: 30 additions & 32 deletions streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
Expand All @@ -60,13 +61,14 @@
import static org.apache.kafka.common.utils.Utils.mkEntry;

public class SmokeTestDriver extends SmokeTestUtil {
private static final String[] TOPICS = new String[] {
private static final String[] TOPICS = {

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Apparently this is unnecessary now?

"data",
"echo",
"max",
"min", "min-suppressed", "min-raw",
"dif",
"sum",
"sws-raw", "sws-suppressed",
"cnt",
"avg",
"tagg"
Expand All @@ -80,18 +82,18 @@ private static class ValueList {
private int index;

ValueList(final int min, final int max) {
this.key = min + "-" + max;
key = min + "-" + max;

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

little bit of cleanup.


this.values = new int[max - min + 1];
for (int i = 0; i < this.values.length; i++) {
this.values[i] = min + i;
values = new int[max - min + 1];
for (int i = 0; i < values.length; i++) {
values[i] = min + i;
}
// We want to randomize the order of data to test not completely predictable processing order
// However, values are also use as a timestamp of the record. (TODO: separate data and timestamp)
// We keep some correlation of time and order. Thus, the shuffling is done with a sliding window
shuffle(this.values, 10);
shuffle(values, 10);

this.index = 0;
index = 0;
}

int next() {
Expand All @@ -106,6 +108,7 @@ public static String[] topics() {
public static Map<String, Set<Integer>> generate(final String kafka,
final int numKeys,
final int maxRecordsPerKey,
final Duration timeToSpend,

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ability to slow down data production so that the system tests can have time to bounce nodes while processing instead of after processing.

I happened to examine the smoke-test system test logs, and realized that we were fully processing the results before the first bounce. This totally circumvents the purpose of the test.

To get the prior behavior, you can just pass Duration.ZERO. The computation below will yield a record pause time of 0, and then the sleep max will produce the prior value of 2ms.

@bbejeck bbejeck Mar 6, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good call!

@guozhangwang guozhangwang Mar 7, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hey @vvcephei I'm wondering if we can simplify the code further: with timeToSpend now can we just remove autoTerminate?

  1. SmokeTestDriverIntegrationTest.java: we are sending 1000 * 10 = 10K records only, so likely that will stop even before the second instance can be started, similar to the issue you observed in system test. I'd suggest we just set timeToSpend to 10 seconds so we have some enough time to start up to 10 / 1 (sleep time) = 10 instances.

  2. As for StreamsSmokeTest itself, we only disableAutoTerminate in three of StreamsUpgradeTest cases, since we do not need to verify the number of records sent / received but only check upgrade completed. We can also use 30 seconds (not sure if it is sufficient, but we can get on average how much time those three tests will take from our nightlies), and even if the test completes before that it will call driver.stop to force-stop the driver anyways.

@guozhangwang guozhangwang Mar 7, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Of course we can then remove the whole disableAutoTerminate thing from StreamsSmokeTest.

@vvcephei vvcephei Mar 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, this is an interesting idea. I'll look into it.

That's a very good point about the integration test, I'll look at that as well.

@vvcephei vvcephei Mar 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, After trying to simplify the code, I found that it was actually better to split the two use cases, so that's what I did. It lets us ditch a whole bunch of bookkeeping when we just want to generate until the process is killed, and it also lets us skip a bunch of mind-bending boolean logic when we do want to just generate a fixed amount of data.

I also added a duration of 20 seconds to the integration test, and it didn't increase the run-time of the test, but now we're getting the same guarantee that we get with the system tests now.

final boolean autoTerminate) {
final Properties producerProps = new Properties();
producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "SmokeTest");
Expand All @@ -131,6 +134,8 @@ public static Map<String, Set<Integer>> generate(final String kafka,
remaining = data.length;
}

final long recordPauseTime = timeToSpend.toMillis() / numKeys / maxRecordsPerKey;

List<ProducerRecord<byte[], byte[]>> needRetry = new ArrayList<>();

while (remaining > 0) {
Expand All @@ -155,9 +160,9 @@ public static Map<String, Set<Integer>> generate(final String kafka,
numRecordsProduced++;
allData.get(key).add(value);
if (numRecordsProduced % 100 == 0) {
System.out.println(numRecordsProduced + " records produced");
System.out.println(Instant.now() + " " + numRecordsProduced + " records produced");
}
Utils.sleep(2);
Utils.sleep(Math.max(recordPauseTime, 2));
}
}
producer.flush();
Expand Down Expand Up @@ -232,12 +237,6 @@ private static void shuffle(final int[] data, @SuppressWarnings("SameParameterVa
}

public static class NumberDeserializer implements Deserializer<Number> {

@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {

}

@Override
public Number deserialize(final String topic, final byte[] data) {
final Number value;
Expand All @@ -247,6 +246,8 @@ public Number deserialize(final String topic, final byte[] data) {
case "min":
case "min-raw":
case "min-suppressed":
case "sws-raw":
case "sws-suppressed":
case "max":
case "dif":
value = intSerde.deserializer().deserialize(topic, data);
Expand All @@ -264,11 +265,6 @@ public Number deserialize(final String topic, final byte[] data) {
}
return value;
}

@Override
public void close() {

}
}

public static VerificationResult verify(final String kafka,
Expand All @@ -279,6 +275,7 @@ public static VerificationResult verify(final String kafka,
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, NumberDeserializer.class);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

final KafkaConsumer<String, Number> consumer = new KafkaConsumer<>(props);
final List<TopicPartition> partitions = getAllPartitions(consumer, TOPICS);
Expand Down Expand Up @@ -406,7 +403,12 @@ private static VerificationResult verifyAll(final Map<String, Set<Integer>> inpu
boolean pass;
try (final PrintStream resultStream = new PrintStream(byteArrayOutputStream)) {
pass = verifyTAgg(resultStream, inputs, events.get("tagg"));
pass &= verifySuppressed(resultStream, "min-suppressed", inputs, events, SmokeTestDriver::getMin);
pass &= verifySuppressed(resultStream, "min-suppressed", events);
pass &= verify(resultStream, "min-suppressed", inputs, events, windowedKey -> {
final String unwindowedKey = windowedKey.substring(1, windowedKey.length() - 1).replaceAll("@.*", "");
return getMin(unwindowedKey);
});

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Decomponsed verifySuppressed and verify, so that we can just verify that sws-suppressed is actually suppressed, without verifying the results (which we would have to compute from first principles)

pass &= verifySuppressed(resultStream, "sws-suppressed", events);
pass &= verify(resultStream, "min", inputs, events, SmokeTestDriver::getMin);
pass &= verify(resultStream, "max", inputs, events, SmokeTestDriver::getMax);
pass &= verify(resultStream, "dif", inputs, events, key -> getMax(key).intValue() - getMin(key).intValue());
Expand Down Expand Up @@ -455,11 +457,10 @@ private static boolean verify(final PrintStream resultStream,
}


@SuppressWarnings("DynamicRegexReplaceableByCompiledPattern")

@guozhangwang guozhangwang Mar 7, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm this seems an IDE specific as well? https://github.com/rtyley/intellij-community-set-committer-to-author/blob/master/plugins/InspectionGadgets/src/inspectionDescriptions/DynamicRegexReplaceableByCompiledPattern.html

@vvcephei vvcephei Mar 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, oops. I'll remove it.

Comment thread
vvcephei marked this conversation as resolved.
Outdated
private static boolean verifySuppressed(final PrintStream resultStream,
@SuppressWarnings("SameParameterValue") final String topic,
final Map<String, Set<Integer>> inputs,
final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events,
final Function<String, Number> getMin) {
final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events) {
resultStream.println("verifying suppressed " + topic);
final Map<String, LinkedList<ConsumerRecord<String, Number>>> topicEvents = events.getOrDefault(topic, emptyMap());
for (final Map.Entry<String, LinkedList<ConsumerRecord<String, Number>>> entry : topicEvents.entrySet()) {
Expand All @@ -476,14 +477,11 @@ private static boolean verifySuppressed(final PrintStream resultStream,
return false;
}
}
return verify(resultStream, topic, inputs, events, windowedKey -> {
final String unwindowedKey = windowedKey.substring(1, windowedKey.length() - 1).replaceAll("@.*", "");
return getMin.apply(unwindowedKey);
});
return true;
}

private static String indent(@SuppressWarnings("SameParameterValue") final String prefix,
final LinkedList<ConsumerRecord<String, Number>> list) {
final Iterable<ConsumerRecord<String, Number>> list) {

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

relaxing type constraints

final StringBuilder stringBuilder = new StringBuilder();
for (final ConsumerRecord<String, Number> record : list) {
stringBuilder.append(prefix).append(record).append('\n');
Expand All @@ -494,13 +492,13 @@ private static String indent(@SuppressWarnings("SameParameterValue") final Strin
private static Long getSum(final String key) {
final int min = getMin(key).intValue();
final int max = getMax(key).intValue();
return ((long) min + (long) max) * (max - min + 1L) / 2L;
return ((long) min + max) * (max - min + 1L) / 2L;

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the second cast is unnecessary to prevent overflow

}

private static Double getAvg(final String key) {
final int min = getMin(key).intValue();
final int max = getMax(key).intValue();
return ((long) min + (long) max) / 2.0;
return ((long) min + max) / 2.0;
}


Expand Down Expand Up @@ -554,7 +552,7 @@ private static Number getMax(final String key) {
}

private static List<TopicPartition> getAllPartitions(final KafkaConsumer<?, ?> consumer, final String... topics) {
final ArrayList<TopicPartition> partitions = new ArrayList<>();
final List<TopicPartition> partitions = new ArrayList<>();

for (final String topic : topics) {
for (final PartitionInfo info : consumer.partitionsFor(topic)) {
Expand Down
20 changes: 15 additions & 5 deletions streams/src/test/java/org/apache/kafka/streams/tests/StreamsSmokeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,14 @@
import org.apache.kafka.streams.StreamsConfig;

import java.io.IOException;
import java.time.Duration;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;

import static org.apache.kafka.streams.tests.SmokeTestDriver.generate;

public class StreamsSmokeTest {

/**
Expand Down Expand Up @@ -62,16 +65,23 @@ public static void main(final String[] args) throws InterruptedException, IOExce
final int numKeys = 10;
final int maxRecordsPerKey = 500;
if (disableAutoTerminate) {
SmokeTestDriver.generate(kafka, numKeys, maxRecordsPerKey, false);
generate(kafka, numKeys, maxRecordsPerKey, Duration.ZERO, false);

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

since this is just going to run until stopped, there's no need (or any sensible value) for delay.

} else {
final Map<String, Set<Integer>> allData = SmokeTestDriver.generate(kafka, numKeys, maxRecordsPerKey, true);
// slow down data production to span 30 seconds so that system tests have time to
// do their bounces, etc.
final Map<String, Set<Integer>> allData =
generate(kafka, numKeys, maxRecordsPerKey, Duration.ofSeconds(30), true);
SmokeTestDriver.verify(kafka, allData, maxRecordsPerKey);
}
break;
case "process":
// this starts a KafkaStreams client
final SmokeTestClient client = new SmokeTestClient(UUID.randomUUID().toString());
client.start(streamsProperties);
// this starts the stream processing app
new SmokeTestClient(UUID.randomUUID().toString()).start(streamsProperties);
break;
case "process-eos":

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added the ability to run the smoke test app with EOS

@guozhangwang guozhangwang Mar 7, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we use StreamsEosTest instead? See my other comment below.

// this starts the stream processing app with EOS
streamsProperties.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
new SmokeTestClient(UUID.randomUUID().toString()).start(streamsProperties);
break;
case "close-deadlock-test":
final ShutdownDeadlockTest test = new ShutdownDeadlockTest(kafka);
Expand Down
4 changes: 4 additions & 0 deletions tests/kafkatest/services/streams.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,6 +353,10 @@ class StreamsSmokeTestJobRunnerService(StreamsSmokeTestBaseService):
def __init__(self, test_context, kafka):
super(StreamsSmokeTestJobRunnerService, self).__init__(test_context, kafka, "process")

class StreamsSmokeTestEOSJobRunnerService(StreamsSmokeTestBaseService):
def __init__(self, test_context, kafka):
super(StreamsSmokeTestEOSJobRunnerService, self).__init__(test_context, kafka, "process-eos")

@vvcephei vvcephei Mar 6, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let me know if there's a better way to do this...

@bbejeck bbejeck Mar 6, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I thought there were already existingEOS tests that use the StreamsSmokeTest?

@guozhangwang guozhangwang Mar 7, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To my other comment above: what's the difference of running StreamsSmokeTestBaseService#process-eos v.s. StreamsEosTestBaseService#process? The former uses StreamsSmokeTest while the latter use StreamsEosTest client. Can we just consolidate them?

@vvcephei vvcephei Mar 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shameful admission: I thought about this, but just didn't take the time to check.

The bounce test was obviously duplicated, but the eos test is more complex. I knew I'd have to do a case-by-case check to make sure I didn't remove coverage. I'll do my homework now...

@vvcephei vvcephei Mar 7, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, so the smoke test and the eos test are similar, but not identical.

The smoke test application has more features in it, though. So, we have more feature coverage under eos when we test with the smoke test.

The eos test evaluates two topologies one with no repartition, and one with a repartition. The smoke test topology contains repartitions, so it only tests with repartition. I think that it should be sufficient to test only with repartition.

The eos test verification specifically checks that "all transactions finished" (org.apache.kafka.streams.tests.EosTestDriver#verifyAllTransactionFinished). I'm not clear on exactly what we're looking for here. It looks like we create a transactional producer and send a record to each partition and then expect to get all those records back, without seeing any other records. But I'm not sure why we're doing this. If we want to drop the eos test, then we might need to add this to the smoke test verification.

WDYT?

@guozhangwang guozhangwang Mar 8, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

verifyAllTransactionFinished aimed to avoid a situation that some dangling txn is open forever, without committing or aborting. Because the consumer needs to guarantee offset ordering when returning data, with read-committed they will also be blocked on those open txn's data forever (this usually indicates a broker-side issue, not streams though, but still).

I think we should still retain this check if we want to merge in the EosTest to SmokeTest, but also if you do not want to drag on it we could keep it as a separate ticket and merge this as-is. Your call :)

@vvcephei vvcephei Mar 8, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, thanks for the explanation. Let's merge this, and I'll create a ticket to follow up by including this check and removing the eos test.

That'll help review burden as well, I think.



class StreamsEosTestDriverService(StreamsEosTestBaseService):
def __init__(self, test_context, kafka):
Expand Down
Loading