This is an automated email from the ASF dual-hosted git repository. mjsax pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push: new 1528264f021 MINOR: update EosIntegrationTest (#16697) 1528264f021 is described below commit 1528264f0217820a86f01964219dbe11b3a60c0e Author: Matthias J. Sax <matth...@confluent.io> AuthorDate: Wed Jul 31 11:25:44 2024 -0700 MINOR: update EosIntegrationTest (#16697) Refactor test to move off deprecated `transform()` in favor of `process()`. Reviewers: Bill Bejeck <b...@confluent.io> --- .../streams/integration/EosIntegrationTest.java | 125 ++++++++++----------- 1 file changed, 59 insertions(+), 66 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java index 38e7e5cc0ca..94e48ee3d49 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java @@ -43,12 +43,10 @@ import org.apache.kafka.streams.errors.TaskCorruptedException; import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster; import org.apache.kafka.streams.integration.utils.IntegrationTestUtils; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.Transformer; -import org.apache.kafka.streams.kstream.TransformerSupplier; -import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.processor.StateRestoreListener; import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.api.Processor; +import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; import org.apache.kafka.streams.processor.internals.StreamThread; import org.apache.kafka.streams.query.QueryResult; @@ -863,8 +861,8 @@ public class EosIntegrationTest { final Properties streamsConfiguration = new Properties(); streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); - streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); - streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); + streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class); + streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class); streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, eosConfig); streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1); streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); @@ -901,10 +899,10 @@ public class EosIntegrationTest { .addSource("source", MULTI_PARTITION_INPUT_TOPIC) .addProcessor("processor", () -> new Processor<Integer, String, Integer, String>() { KeyValueStore<Integer, String> stateStore; - org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context; + ProcessorContext<Integer, String> context; @Override - public void init(final org.apache.kafka.streams.processor.api.ProcessorContext<Integer, String> context) { + public void init(final ProcessorContext<Integer, String> context) { Processor.super.init(context); this.context = context; stateStore = context.getStateStore(stateStoreName); @@ -1014,7 +1012,7 @@ public class EosIntegrationTest { final long topicEndOffset = consumer.position(tp); assertTrue(topicEndOffset >= checkpointedOffset, - "changelog topic end " + topicEndOffset + " is less than checkpointed offset " + checkpointedOffset); + "changelog topic end " + topicEndOffset + " is less than checkpointed offset " + checkpointedOffset); consumer.seekToBeginning(partitions); @@ -1070,80 +1068,75 @@ public class EosIntegrationTest { } final KStream<Long, Long> input = builder.stream(MULTI_PARTITION_INPUT_TOPIC); - input.transform(new TransformerSupplier<Long, Long, KeyValue<Long, Long>>() { - @SuppressWarnings("unchecked") - @Override - public Transformer<Long, Long, KeyValue<Long, Long>> get() { - return new Transformer<Long, Long, KeyValue<Long, Long>>() { - ProcessorContext context; - KeyValueStore<Long, Long> state = null; + input.process(() -> new Processor<Long, Long, Long, Long>() { + ProcessorContext<Long, Long> context; + KeyValueStore<Long, Long> state = null; - @Override - public void init(final ProcessorContext context) { - this.context = context; + @Override + public void init(final ProcessorContext<Long, Long> context) { + this.context = context; - if (withState) { - state = context.getStateStore(storeName); - } + if (withState) { + state = context.getStateStore(storeName); } + } - @Override - public KeyValue<Long, Long> transform(final Long key, final Long value) { - if (stallInjected.compareAndSet(true, false)) { - LOG.info(dummyHostName + " is executing the injected stall"); - stallingHost.set(dummyHostName); - while (doStall) { - final Thread thread = Thread.currentThread(); - if (thread.isInterrupted()) { + @Override + public void process(final Record<Long, Long> record) { + if (stallInjected.compareAndSet(true, false)) { + LOG.info(dummyHostName + " is executing the injected stall"); + stallingHost.set(dummyHostName); + while (doStall) { + final Thread thread = Thread.currentThread(); + if (thread.isInterrupted()) { + throw new RuntimeException("Detected we've been interrupted."); + } + if (!processingThreadsEnabled) { + if (!((StreamThread) thread).isRunning()) { throw new RuntimeException("Detected we've been interrupted."); } - if (!processingThreadsEnabled) { - if (!((StreamThread) thread).isRunning()) { - throw new RuntimeException("Detected we've been interrupted."); - } - } - try { - Thread.sleep(100); - } catch (final InterruptedException e) { - throw new RuntimeException(e); - } } - } - - if ((value + 1) % 10 == 0) { - context.commit(); - commitRequested.incrementAndGet(); - } - - if (state != null) { - Long sum = state.get(key); - - if (sum == null) { - sum = value; - } else { - sum += value; + try { + Thread.sleep(100); + } catch (final InterruptedException e) { + throw new RuntimeException(e); } - state.put(key, sum); - state.flush(); } + } + final long key = record.key(); + final long value = record.value(); - if (errorInjected.compareAndSet(true, false)) { - // only tries to fail once on one of the task - throw new RuntimeException("Injected test exception."); - } + if ((value + 1) % 10 == 0) { + context.commit(); + commitRequested.incrementAndGet(); + } + + if (state != null) { + Long sum = state.get(key); - if (state != null) { - return new KeyValue<>(key, state.get(key)); + if (sum == null) { + sum = value; } else { - return new KeyValue<>(key, value); + sum += value; } + state.put(key, sum); + state.flush(); } - @Override - public void close() { } - }; - } }, storeNames) + + if (errorInjected.compareAndSet(true, false)) { + // only tries to fail once on one of the task + throw new RuntimeException("Injected test exception."); + } + + if (state != null) { + context.forward(record.withValue(state.get(key))); + } else { + context.forward(record); + } + } + }, storeNames) .to(SINGLE_PARTITION_OUTPUT_TOPIC); stateTmpDir = TestUtils.tempDirectory().getPath() + File.separator;