Skip to content

KAFKA-14911: Add system tests for rolling upgrade path of KIP-904 #13656

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

Closed
mjsax wants to merge 3 commits into from
Closed

KAFKA-14911: Add system tests for rolling upgrade path of KIP-904 #13656

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
adc67d5
add system test
vcrfxia Apr 25, 2023
c6d9447
- add table-aggregation to older versions
mjsax May 2, 2023
a70e6ba
Github comments
mjsax May 5, 2023
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
10 changes: 7 additions & 3 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 @@ -106,7 +106,11 @@ private static class ValueList {
}

int next() {
return (index < values.length) ? values[index++] : -1;

@mjsax mjsax May 2, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

The test produces too much data, and thus ended up sending -1 for a lot of records at the end.

final int v = values[index++];
if (index >= values.length) {
index = 0;
}
Comment on lines +108 to +110

@cadonna cadonna May 3, 2023

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.

Doesn't this risk to bring a lot of disorder into the timestamps? I am referring to the comment on line 100. What are the consequences of such a disorder?

@mjsax mjsax May 5, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Seems the comment is outdated. The custom TS-extractor was removed years ago: 52e3979

Let me delete the comment.

return v;
}
}

Expand All @@ -126,7 +130,7 @@ static void generatePerpetually(final String kafka,
data[i] = new ValueList(i, i + maxRecordsPerKey - 1);
}

final Random rand = new Random();
final Random rand = new Random(System.currentTimeMillis());

@mjsax mjsax May 2, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Minor side improvement


try (final KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(producerProps)) {
while (true) {
Expand Down Expand Up @@ -313,7 +317,7 @@ public void onCompletion(final RecordMetadata metadata, final Exception exceptio
}

private static void shuffle(final int[] data, @SuppressWarnings("SameParameterValue") final int windowSize) {
final Random rand = new Random();
final Random rand = new Random(System.currentTimeMillis());
for (int i = 0; i < data.length; i++) {
// we shuffle data within windowSize
final int j = rand.nextInt(Math.min(data.length - i, windowSize)) + i;
Expand Down
24 changes: 24 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,10 +34,34 @@ public class SmokeTestUtil {

final static int END = Integer.MAX_VALUE;

static ProcessorSupplier<Object, Object, Void, Void> printTaskProcessorSupplier(final String topic) {
return printTaskProcessorSupplier(topic, "");
}

static ProcessorSupplier<Object, Object, Void, Void> printProcessorSupplier(final String topic) {
return printProcessorSupplier(topic, "");
}

static ProcessorSupplier<Object, Object, Void, Void> printTaskProcessorSupplier(final String topic, final String name) {

@vcrfxia vcrfxia May 2, 2023

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.

nit: this new processor is the same as the existing one except that it doesn't track or print the number of records processed, right? Would it be better to have a boolean to toggle the print behavior, rather than duplicating the rest of the processor code? (Not a big deal either way since it's not much code, but as a reader I had to spent some time determining/verifying that the print behavior is the only difference.)

@mjsax mjsax May 2, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Yeah, it was a quick-and-dirty thing -- guess it might make sense to actually have a single Processor and let the original one inherit from the new one.

@cadonna cadonna May 3, 2023

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.

It seems the parameter name is not used.

return () -> new ContextualProcessor<Object, Object, Void, Void>() {
@Override
public void init(final ProcessorContext<Void, Void> context) {
super.init(context);
System.out.println("[DEV] initializing processor: topic=" + topic + " taskId=" + context.taskId());
System.out.flush();
}

@Override
public void process(final Record<Object, Object> record) { }

@Override
public void close() {
System.out.printf("Close processor for task %s%n", context().taskId());
System.out.flush();
}
};
}

static ProcessorSupplier<Object, Object, Void, Void> printProcessorSupplier(final String topic, final String name) {
return () -> new ContextualProcessor<Object, Object, Void, Void>() {
private int numRecordsProcessed = 0;
Expand Down
118 changes: 118 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@
*/
package org.apache.kafka.streams.tests;

import java.util.Arrays;
import java.util.Collections;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
Expand All @@ -24,18 +26,24 @@
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.utils.ByteBufferInputStream;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KafkaClientSupplier;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.TaskAssignmentException;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
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.processor.TaskId;
import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
Expand Down Expand Up @@ -69,6 +77,7 @@
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION;
import static org.apache.kafka.streams.tests.SmokeTestUtil.intSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.longSerde;
import static org.apache.kafka.streams.tests.SmokeTestUtil.stringSerde;

public class StreamsUpgradeTest {
Expand All @@ -84,6 +93,7 @@ public static void main(final String[] args) throws Exception {
System.out.println("StreamsTest instance started (StreamsUpgradeTest trunk)");
System.out.println("props=" + streamsProperties);

// Do not use try-with-resources here; otherwise KafkaStreams will be closed when `main()` exits
final KafkaStreams streams = buildStreams(streamsProperties);
streams.start();

Expand All @@ -107,6 +117,9 @@ public static KafkaStreams buildStreams(final Properties streamsProperties) {
final boolean runFkJoin = Boolean.parseBoolean(streamsProperties.getProperty(
"test.run_fk_join",
"false"));
final boolean runTableAgg = Boolean.parseBoolean(streamsProperties.getProperty(
"test.run_table_agg",
"false"));
if (runFkJoin) {
try {
final KTable<Integer, String> fkTable = builder.table(
Expand All @@ -116,6 +129,23 @@ public static KafkaStreams buildStreams(final Properties streamsProperties) {
System.err.println("Caught " + e.getMessage());
}
}
if (runTableAgg) {
final String aggProduceValue = streamsProperties.getProperty("test.agg_produce_value", "");
if (aggProduceValue.isEmpty()) {
System.err.printf("'%s' must be specified when '%s' is true.", "test.agg_produce_value", "test.run_table_agg");
}
final String expectedAggValuesStr = streamsProperties.getProperty("test.expected_agg_values", "");
if (expectedAggValuesStr.isEmpty()) {
System.err.printf("'%s' must be specified when '%s' is true.", "test.expected_agg_values", "test.run_table_agg");
}
final List<String> expectedAggValue = Arrays.asList(expectedAggValuesStr.split(","));

try {
buildTableAgg(dataTable, aggProduceValue, expectedAggValue);
} catch (final Exception e) {
System.err.println("Caught " + e.getMessage());
}
}

final Properties config = new Properties();
config.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "StreamsUpgradeTest");
Expand Down Expand Up @@ -143,6 +173,94 @@ private static void buildFKTable(final KStream<String, Integer> primaryTable,
kStream.to("fk-result", Produced.with(stringSerde, stringSerde));
}

private static void buildTableAgg(final KTable<String, Integer> sourceTable,
final String aggProduceValue,
final List<String> expectedAggValues) {
final KStream<Integer, String> result = sourceTable
.groupBy(
(k, v) -> new KeyValue<>(v, aggProduceValue),

@mjsax mjsax May 2, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Changed this to use v as key -- works just fine

Grouped.with(intSerde, stringSerde))
.aggregate(
() -> new Agg(Collections.emptyList(), 0),
(k, v, agg) -> {
final List<String> seenValues;
final boolean updated;
if (!agg.seenValues.contains(v)) {
seenValues = new ArrayList<>(agg.seenValues);
seenValues.add(v);
Collections.sort(seenValues);

@fqaiser94 fqaiser94 May 10, 2023

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.

Is there a reason we need to sort?

updated = true;
} else {
seenValues = agg.seenValues;
updated = false;
}

final boolean shouldLog = updated || (agg.recordsProcessed % 10 == 0); // value of 10 is chosen for debugging purposes. can increase to 100 once test is passing.
if (shouldLog) {

@mjsax mjsax May 2, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Changed this slightly to avoid spamming the output

@vcrfxia vcrfxia May 2, 2023

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 I'm not seeing what the change was. Should we increase the value in the line above from 10 too 100? Currently the comment still says "value of 10 is chosen for debugging purposes. can increase to 100 once test is passing"

@mjsax mjsax May 2, 2023 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Originally it was

if (shouldLog && seenValues.containsAll(expectedAggValues) {
 ...
} else {
 ...
}

So it always logged something.

@vcrfxia vcrfxia May 2, 2023

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.

Ah, must've put that in for debugging and forgotten to leave a note. The new mechanism makes sense 👍 If you think recordsProcessed % 10 is the right frequency, rather than recordsProcessed % 100, then perhaps we can remove the comment entirely and just leave it as recordsProcessed % 10.

if (seenValues.containsAll(expectedAggValues)) {
System.out.printf("Table aggregate processor saw expected values. Seen: %s. Expected: %s%n", String.join(",", seenValues), String.join(",", expectedAggValues));
} else {
System.out.printf("Table aggregate processor did not see expected values. Seen: %s. Expected: %s%n", String.join(",", seenValues), String.join(",", expectedAggValues)); // this line for debugging purposes only.
}
}

return new Agg(seenValues, agg.recordsProcessed + 1);
},
(k, v, agg) -> agg,
Materialized.with(intSerde, new AggSerde()))
.mapValues((k, vAgg) -> String.join(",", vAgg.seenValues))
.toStream();

// adding dummy processor for better debugging (will print assigned tasks)
result.process(SmokeTestUtil.printTaskProcessorSupplier("table-repartition"));
result.to("table-agg-result", Produced.with(intSerde, stringSerde));
}

private static class Agg {
private final List<String> seenValues;
private final long recordsProcessed;

Agg(final List<String> seenValues, final long recordsProcessed) {
this.seenValues = seenValues;
this.recordsProcessed = recordsProcessed;
}

byte[] serialize() {
final byte[] rawSeenValues = stringSerde.serializer().serialize("", String.join(",", seenValues));
final byte[] rawRecordsProcessed = longSerde.serializer().serialize("", recordsProcessed);
return ByteBuffer
.allocate(rawSeenValues.length + rawRecordsProcessed.length)
.put(rawSeenValues)
.put(rawRecordsProcessed)
.array();
}

static Agg deserialize(final byte[] rawAgg) {
final byte[] rawSeenValues = new byte[rawAgg.length - 8];
System.arraycopy(rawAgg, 0, rawSeenValues, 0, rawSeenValues.length);
final List<String> seenValues = Arrays.asList(stringSerde.deserializer().deserialize("", rawSeenValues).split(","));

final byte[] rawRecordsProcessed = new byte[8];
System.arraycopy(rawAgg, rawAgg.length - 8, rawRecordsProcessed, 0, 8);
final long recordsProcessed = longSerde.deserializer().deserialize("", rawRecordsProcessed);

return new Agg(seenValues, recordsProcessed);
}
}

private static class AggSerde implements Serde<Agg> {

@Override
public Serializer<Agg> serializer() {
return (topic, agg) -> agg.serialize();
}

@Override
public Deserializer<Agg> deserializer() {
return (topic, rawAgg) -> Agg.deserialize(rawAgg);
}
}

private static class FutureKafkaClientSupplier extends DefaultKafkaClientSupplier {
@Override
public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
Expand Down
10 changes: 7 additions & 3 deletions ...upgrade-system-tests-33/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 @@ -106,7 +106,11 @@ private static class ValueList {
}

int next() {
return (index < values.length) ? values[index++] : -1;
final int v = values[index++];
if (index >= values.length) {
index = 0;
}
return v;
}
}

Expand All @@ -126,7 +130,7 @@ static void generatePerpetually(final String kafka,
data[i] = new ValueList(i, i + maxRecordsPerKey - 1);
}

final Random rand = new Random();
final Random rand = new Random(System.currentTimeMillis());

try (final KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(producerProps)) {
while (true) {
Expand Down Expand Up @@ -309,7 +313,7 @@ public void onCompletion(final RecordMetadata metadata, final Exception exceptio
}

private static void shuffle(final int[] data, @SuppressWarnings("SameParameterValue") final int windowSize) {
final Random rand = new Random();
final Random rand = new Random(System.currentTimeMillis());
for (int i = 0; i < data.length; i++) {
// we shuffle data within windowSize
final int j = rand.nextInt(Math.min(data.length - i, windowSize)) + i;
Expand Down
24 changes: 24 additions & 0 deletions ...s/upgrade-system-tests-33/src/test/java/org/apache/kafka/streams/tests/SmokeTestUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,10 +34,34 @@ public class SmokeTestUtil {

final static int END = Integer.MAX_VALUE;

static ProcessorSupplier<Object, Object, Void, Void> printTaskProcessorSupplier(final String topic) {
return printTaskProcessorSupplier(topic, "");
}

static ProcessorSupplier<Object, Object, Void, Void> printProcessorSupplier(final String topic) {
return printProcessorSupplier(topic, "");
}

static ProcessorSupplier<Object, Object, Void, Void> printTaskProcessorSupplier(final String topic, final String name) {
return () -> new ContextualProcessor<Object, Object, Void, Void>() {
@Override
public void init(final ProcessorContext<Void, Void> context) {
super.init(context);
System.out.println("[3.3] initializing processor: topic=" + topic + " taskId=" + context.taskId());
System.out.flush();
}

@Override
public void process(final Record<Object, Object> record) {}

@Override
public void close() {
System.out.printf("Close processor for task %s%n", context().taskId());
System.out.flush();
}
};
}

static ProcessorSupplier<Object, Object, Void, Void> printProcessorSupplier(final String topic, final String name) {
return () -> new ContextualProcessor<Object, Object, Void, Void>() {
private int numRecordsProcessed = 0;
Expand Down
Loading