SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Kafka Streams binder test cleanup

Browse Source
This commit is contained in:
Soby Chacko
2023-12-08 17:32:14 -05:00
parent 4d49a27cc1
commit 95775b5eba
5 changed files with 25 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java
View File

@@ -441,7 +441,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofSeconds(5))).count(Materialized.as("foobar-WordCounts"))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5))).count(Materialized.as("foobar-WordCounts"))
.toStream()
.map((key, value) -> new KeyValue<>(null, null));
}

1
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java
View File

@@ -328,6 +328,7 @@ class KafkaStreamsComponentBeansTests {
KStream<String, String>[]> {
@Override
@SuppressWarnings("unchecked")
public KStream<String, String>[] apply(KStream<Integer, String> stringIntegerKStream) {
return stringIntegerKStream.map((key, value) -> new KeyValue<>(key.toString(), value))
.split()

21
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java
View File

@@ -22,9 +22,8 @@ import java.util.Map;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
@@ -121,16 +120,16 @@ class KafkaStreamsFunctionStateStoreTests {
@Bean(name = "biConsumerBean")
public java.util.function.BiConsumer<KStream<Object, String>, KStream<Object, String>> process() {
return (input0, input1) ->
input0.process((ProcessorSupplier<Object, String>) () -> new Processor<Object, String>() {
input0.process(() -> new Processor<Object, String, Object, String>() {
@Override
@SuppressWarnings("unchecked")
public void init(ProcessorContext context) {
public void init(org.apache.kafka.streams.processor.api.ProcessorContext<Object, String> context) {
state1 = (KeyValueStore<Long, Long>) context.getStateStore("my-store");
state2 = (WindowStore<Long, Long>) context.getStateStore("other-store");
}
@Override
public void process(Object key, String value) {
public void process(Record<Object, String> record) {
processed1 = true;
}
@@ -144,16 +143,16 @@ class KafkaStreamsFunctionStateStoreTests {
@Bean
public java.util.function.Consumer<KTable<Object, String>> hello() {
return input -> {
input.toStream().process(() -> new Processor<Object, String>() {
input.toStream().process(() -> new Processor<Object, String, Object, String>() {
@Override
@SuppressWarnings("unchecked")
public void init(ProcessorContext context) {
public void init(org.apache.kafka.streams.processor.api.ProcessorContext<Object, String> context) {
state3 = (KeyValueStore<Long, Long>) context.getStateStore("my-store");
state4 = (WindowStore<Long, Long>) context.getStateStore("other-store");
}
@Override
public void process(Object key, String value) {
public void process(Record<Object, String> record) {
processed2 = true;
}

21
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java
View File

@@ -24,8 +24,8 @@ import java.util.function.BiConsumer;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.Record;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
@@ -143,13 +143,15 @@ class KafkaStreamsRetryTests {
public java.util.function.Consumer<KStream<Object, String>> process(@Lazy @Qualifier("process-in-0-RetryTemplate") RetryTemplate retryTemplate) {
return input -> input
.process(() -> new Processor<Object, String>() {
.process(() -> new Processor<Object, String, Object, String>() {
@Override
public void init(ProcessorContext processorContext) {
public void init(org.apache.kafka.streams.processor.api.ProcessorContext<Object, String> context) {
Processor.super.init(context);
}
@Override
public void process(Object o, String s) {
public void process(Record<Object, String> record) {
retryTemplate.execute(context -> {
LATCH1.countDown();
throw new RuntimeException();
@@ -191,18 +193,19 @@ class KafkaStreamsRetryTests {
public java.util.function.Consumer<KStream<Object, String>> process() {
return input -> input
.process(() -> new Processor<Object, String>() {
.process(() -> new Processor<Object, String, Object, String>() {
@Override
public void init(ProcessorContext processorContext) {
public void init(org.apache.kafka.streams.processor.api.ProcessorContext<Object, String> context) {
Processor.super.init(context);
}
@Override
public void process(Object o, String s) {
public void process(Record<Object, String> record) {
fooRetryTemplate().execute(context -> {
LATCH2.countDown();
throw new RuntimeException();
});
}
@Override

2
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java
View File

@@ -562,7 +562,7 @@ class StreamToTableJoinFunctionTests {
return (input1Stream, input2Stream) -> input1Stream
.join(input2Stream,
(event1, event2) -> null,
JoinWindows.of(Duration.ofMillis(5)),
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMillis(5)),
StreamJoined.with(
Serdes.String(),
Serdes.String(),