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

Kafka binders test migration to Kafka 4.0.0 client

Browse Source 
- Replace deprecated ValueTransformerWithKey with FixedKeyProcessor
- Update TimeWindows.of() to TimeWindows.ofSizeWithNoGrace()
- Migrate branch() to split().branch() with Named and Branched
- Replace deprecated KafkaTestUtils.consumerProps() signature
- Update StreamPartitioner return type to Optional<Set<Integer>>
- Modernize Processor API imports and method signatures
- Add EmbeddedKafkaBroker parameter injection for JUnit 5
- Fix Consumer.poll() to use Duration parameter
- Remove unused imports and deprecated API usage

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
This commit is contained in:
Soby Chacko
2025-06-11 20:16:21 -04:00
parent 4d4e64b415
commit 3a937debfa
13 changed files with 171 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java
View File

@@ -24,8 +24,8 @@ import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.errors.UnknownStateStoreException;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.api.FixedKeyProcessor;
import org.apache.kafka.streams.processor.api.FixedKeyRecord;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.StoreBuilder;
@@ -93,11 +93,9 @@ class InteractiveQueryServiceMultiStateStoreTests {
.web(WebApplicationType.NONE)
.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.function.definition=app1;app2",
"--spring.cloud.function.definition=app1",
"--spring.cloud.stream.function.bindings.app1-in-0=input1",
"--spring.cloud.stream.function.bindings.app2-in-0=input2",
"--spring.cloud.stream.kafka.streams.binder.functions.app1.application-id=stateStoreTestApp1",
"--spring.cloud.stream.kafka.streams.binder.functions.app2.application-id=stateStoreTestApp2",
appServerArg,
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())
) {
@@ -111,8 +109,6 @@ class InteractiveQueryServiceMultiStateStoreTests {
for (int i = 0; i < 100; i++) {
assertThat(queryService.getQueryableStore(STORE_1_NAME, QueryableStoreTypes.keyValueStore())
.get("someKey")).isNull();
assertThat(queryService.getQueryableStore(STORE_2_NAME, QueryableStoreTypes.keyValueStore())
.get("someKey")).isNull();
}
}
}
@@ -203,7 +199,7 @@ class InteractiveQueryServiceMultiStateStoreTests {
@Bean
public Consumer<KStream<String, String>> app1() {
return s -> s
.transformValues(EchoTransformer::new, STORE_1_NAME)
.processValues(EchoProcessor::new, STORE_1_NAME)
.foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME));
}
@@ -216,7 +212,7 @@ class InteractiveQueryServiceMultiStateStoreTests {
@Bean
public Consumer<KStream<String, String>> app2() {
return s -> s
.transformValues(EchoTransformer::new, STORE_2_NAME)
.processValues(EchoProcessor::new, STORE_1_NAME)
.foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_2_NAME));
}
@@ -286,7 +282,7 @@ class InteractiveQueryServiceMultiStateStoreTests {
@Bean
public Consumer<KStream<String, String>> app1() {
return s -> s
.transformValues(EchoTransformer::new, STORE_1_NAME)
.processValues(EchoProcessor::new, STORE_1_NAME)
.foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME));
}
@@ -301,15 +297,12 @@ class InteractiveQueryServiceMultiStateStoreTests {
}
}
static class EchoTransformer implements ValueTransformerWithKey<String, String, String> {
static class EchoProcessor implements FixedKeyProcessor<String, String, String> {
@Override
public void init(ProcessorContext context) {
}
public void process(FixedKeyRecord<String, String> fixedKeyRecord) {
@Override
public String transform(String key, String value) {
return value;
}
@Override

28
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.java
View File

@@ -29,7 +29,9 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Branched;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Named;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -65,8 +67,7 @@ class MultipleFunctionsInSameAppTests {
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("purchase-groups", "false",
embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "purchase-groups", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
@@ -80,7 +81,7 @@ class MultipleFunctionsInSameAppTests {
@Test
@SuppressWarnings("unchecked")
void testMultiFunctionsInSameApp() throws InterruptedException {
void testMultiFunctionsInSameApp(EmbeddedKafkaBroker embeddedKafka) throws InterruptedException {
SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -103,7 +104,7 @@ class MultipleFunctionsInSameAppTests {
"--spring.cloud.stream.kafka.streams.binder.functions.analyze.configuration.client.id=analyze-client",
"--spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.configuration.client.id=anotherProcess-client",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
receiveAndValidate("purchases", "coffee", "electronics");
receiveAndValidate(embeddedKafka, "purchases", "coffee", "electronics");
StreamsBuilderFactoryBean processStreamsBuilderFactoryBean = context
.getBean("&stream-builder-processItem", StreamsBuilderFactoryBean.class);
@@ -138,7 +139,7 @@ class MultipleFunctionsInSameAppTests {
}
@Test
void testMultiFunctionsInSameAppWithMultiBinders() throws Exception {
void testMultiFunctionsInSameAppWithMultiBinders(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -168,7 +169,7 @@ class MultipleFunctionsInSameAppTests {
"--spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.configuration.client.id=analyze-client")) {
Thread.sleep(1000);
receiveAndValidate("purchases", "coffee", "electronics");
receiveAndValidate(embeddedKafka, "purchases", "coffee", "electronics");
StreamsBuilderFactoryBean processStreamsBuilderFactoryBean = context
.getBean("&stream-builder-processItem", StreamsBuilderFactoryBean.class);
@@ -192,7 +193,7 @@ class MultipleFunctionsInSameAppTests {
}
}
private void receiveAndValidate(String in, String... out) throws InterruptedException {
private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka, String in, String... out) throws InterruptedException {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
@@ -218,9 +219,16 @@ class MultipleFunctionsInSameAppTests {
@Bean
public Function<KStream<String, String>, KStream<String, String>[]> processItem() {
return input -> input.branch(
(s, p) -> p.equalsIgnoreCase("coffee"),
(s, p) -> p.equalsIgnoreCase("electronics"));
return input -> {
Map<String, KStream<String, String>> branches = input.split(Named.as("split-"))
.branch((s, p) -> p.equalsIgnoreCase("coffee"), Branched.as("coffee"))
.branch((s, p) -> p.equalsIgnoreCase("electronics"), Branched.as("electronics"))
.defaultBranch(Branched.as("other"));
return new KStream[] {
branches.get("split-coffee"),
branches.get("split-electronics")
};
};
}
// Testing for the scenario under https://github.com/spring-cloud/spring-cloud-stream/issues/2817

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

@@ -63,8 +63,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests {
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("groupx", "false",
embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "groupx", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
@@ -77,7 +76,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests {
}
@Test
void kstreamWordCountWithStringInputAndPojoOuput() throws Exception {
void kstreamWordCountWithStringInputAndPojoOuput(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -100,14 +99,14 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests {
"=KafkaStreamsBinderWordCountBranchesFunctionTests-abc",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString());
try {
receiveAndValidate(context);
receiveAndValidate(embeddedKafka);
}
finally {
context.close();
}
}
private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception {
private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka) throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
@@ -193,7 +192,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests {
final Map<String, KStream<Object, WordCount>> stringKStreamMap = input
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.groupBy((key, value) -> value)
.windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5)))
.count(Materialized.as("WordCounts-branch"))
.toStream()
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value,

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.function;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Locale;
@@ -86,8 +87,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
@@ -102,7 +102,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
@Test
@SuppressWarnings("unchecked")
void basicKStreamTopologyExecution() throws Exception {
void basicKStreamTopologyExecution(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -123,7 +123,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.consumedAs=custom-consumer",
"--spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.producedAs=custom-producer",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
receiveAndValidate("words", "counts");
receiveAndValidate(embeddedKafka, "words", "counts");
final MeterRegistry meterRegistry = context.getBean(MeterRegistry.class);
Thread.sleep(100);
@@ -179,7 +179,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
}
@Test
void kstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer() {
void kstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -195,12 +195,12 @@ class KafkaStreamsBinderWordCountFunctionTests {
+ "=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.binder.brokers="
+ embeddedKafka.getBrokersAsString())) {
receiveAndValidate("words-5", "counts-5");
receiveAndValidate(embeddedKafka, "words-5", "counts-5");
}
}
@Test
void kstreamWordCountFunctionWithCustomProducerStreamPartitioner() throws Exception {
void kstreamWordCountFunctionWithCustomProducerStreamPartitioner(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -240,7 +240,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
}
@Test
void kstreamBinderAutoStartup() throws Exception {
void kstreamBinderAutoStartup(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -262,7 +262,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
}
@Test
void kstreamIndividualBindingAutoStartup() throws Exception {
void kstreamIndividualBindingAutoStartup(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -287,8 +287,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
// The following test verifies the fixes made for this issue:
// https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/774
@Test
void outboundNullValueIsHandledGracefully()
throws Exception {
void outboundNullValueIsHandledGracefully(EmbeddedKafkaBroker embeddedKafka) {
SpringApplication app = new SpringApplication(OutboundNullApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -323,7 +322,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
}
}
private void receiveAndValidate(String in, String out) {
private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka, String in, String out) {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
@@ -401,7 +400,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofMillis(5000)))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000)))
.count(Materialized.as("foo-WordCounts"))
.toStream()
.map((key, value) -> new KeyValue<>(key.key(), new WordCount(key.key(), value,
@@ -429,7 +428,14 @@ class KafkaStreamsBinderWordCountFunctionTests {
@Bean
StreamPartitioner<String, WordCount> streamPartitioner() {
return (t, k, v, n) -> k.equals("foo") ? 0 : 1;
return (topic, key, value, numPartitions) -> {
if (key.equals("foo")) {
return Optional.of(Collections.singleton(0));
}
else {
return Optional.of(Collections.singleton(1));
}
};
}
}

16
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java
View File

@@ -31,7 +31,6 @@ import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
@@ -46,7 +45,6 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -58,15 +56,8 @@ import static org.assertj.core.api.Assertions.assertThat;
@EmbeddedKafka(topics = {"topic1-dlq", "topic2-dlq"})
class DlqDestinationResolverTests {
private static EmbeddedKafkaBroker embeddedKafka;
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
}
@Test
void dlqDestinationResolverWorks() throws Exception {
void dlqDestinationResolverWorks(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -92,8 +83,7 @@ class DlqDestinationResolverTests {
template.setDefaultTopic("word2");
template.sendDefault("foobar");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("some-random-group",
"false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "some-random-group", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerProps);
@@ -128,7 +118,7 @@ class DlqDestinationResolverTests {
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).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("foo-WordCounts-x"))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x"))
.toStream().map((key, value) -> new KeyValue<>(null,
"Count for " + key.key() + " : " + value));
}

11
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java
View File

@@ -63,8 +63,7 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests {
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id",
"false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group-id", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put("value.deserializer", LongDeserializer.class);
DefaultKafkaConsumerFactory<Integer, Long> cf = new DefaultKafkaConsumerFactory<>(
@@ -79,7 +78,7 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests {
}
@Test
void kstreamBinderWithPojoInputAndStringOuput() throws Exception {
void kstreamBinderWithPojoInputAndStringOuput(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(ProductCountApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
ConfigurableApplicationContext context = app.run("--server.port=0",
@@ -98,14 +97,14 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests {
"--spring.cloud.stream.kafka.streams.binder.brokers="
+ embeddedKafka.getBrokersAsString());
try {
receiveAndValidateFoo();
receiveAndValidateFoo(embeddedKafka);
}
finally {
context.close();
}
}
private void receiveAndValidateFoo() {
private void receiveAndValidateFoo(EmbeddedKafkaBroker embeddedKafka) throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
@@ -128,7 +127,7 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests {
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(new JsonSerde<>(Product.class),
new JsonSerde<>(Product.class)))
.windowedBy(TimeWindows.of(Duration.ofMillis(5000)))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000)))
.count(Materialized.as("id-count-store-x")).toStream()
.map((key, value) -> new KeyValue<>(key.key().id, value));
}

16
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderTombstoneTests.java
View File

@@ -73,8 +73,7 @@ class KafkaStreamsBinderTombstoneTests {
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerProps);
@@ -88,8 +87,7 @@ class KafkaStreamsBinderTombstoneTests {
}
@Test
void sendToTombstone()
throws Exception {
void sendToTombstone(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(
WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@@ -108,7 +106,7 @@ class KafkaStreamsBinderTombstoneTests {
"--spring.cloud.stream.bindings.process-in-0.consumer.concurrency=2",
"--spring.cloud.stream.kafka.streams.binder.brokers="
+ embeddedKafka.getBrokersAsString())) {
receiveAndValidate("words-1", "counts-1");
receiveAndValidate(embeddedKafka, "words-1", "counts-1");
// Assertions on StreamBuilderFactoryBean
StreamsBuilderFactoryBean streamsBuilderFactoryBean = context
.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
@@ -121,7 +119,7 @@ class KafkaStreamsBinderTombstoneTests {
.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG);
assertThat(concurrency).isEqualTo(2);
sendTombStoneRecordsAndVerifyGracefulHandling();
sendTombStoneRecordsAndVerifyGracefulHandling(embeddedKafka);
CleanupConfig cleanup = TestUtils.getPropertyValue(streamsBuilderFactoryBean,
"cleanupConfig", CleanupConfig.class);
@@ -130,7 +128,7 @@ class KafkaStreamsBinderTombstoneTests {
}
}
private void receiveAndValidate(String in, String out) {
private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka, String in, String out) {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
@@ -147,7 +145,7 @@ class KafkaStreamsBinderTombstoneTests {
}
}
private void sendTombStoneRecordsAndVerifyGracefulHandling() throws Exception {
private void sendTombStoneRecordsAndVerifyGracefulHandling(EmbeddedKafkaBroker embeddedKafka) throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
@@ -177,7 +175,7 @@ class KafkaStreamsBinderTombstoneTests {
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofMillis(5000)))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000)))
.count(Materialized.as("foo-WordCounts"))
.toStream()
.map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value,

119
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingDisabledTests.java
View File

@@ -34,64 +34,41 @@ import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.verify;
/**
* @author Soby Chacko
*/
@RunWith(SpringRunner.class)
@ContextConfiguration
@DirtiesContext
public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests {
@EmbeddedKafka(topics = {"decode-counts", "decode-counts-1"}, partitions = 1)
class KafkaStreamsNativeEncodingDecodingDisabledTests {
/**
* Kafka rule.
*/
@ClassRule
public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true,
"decode-counts", "decode-counts-1");
private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule
.getEmbeddedKafka();
@SpyBean
org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsMessageConversionDelegate conversionDelegate;
private static EmbeddedKafkaBroker embeddedKafka;
private static Consumer<String, String> consumer;
@BeforeClass
@BeforeAll
public static void setUp() {
System.setProperty("spring.cloud.stream.kafka.streams.binder.brokers",
embeddedKafka.getBrokersAsString());
System.setProperty("server.port", "0");
System.setProperty("spring.jmx.enabled", "false");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerProps);
@@ -99,51 +76,48 @@ public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests {
embeddedKafka.consumeFromEmbeddedTopics(consumer, "decode-counts", "decode-counts-1");
}
@AfterClass
@AfterAll
public static void tearDown() {
consumer.close();
System.clearProperty("spring.cloud.stream.kafka.streams.binder.brokers");
System.clearProperty("server.port");
System.clearProperty("spring.jmx.enabled");
}
@SpringBootTest(classes = WordCountProcessorApplication.class, webEnvironment = SpringBootTest.WebEnvironment.NONE,
properties = {
"spring.cloud.stream.bindings.process-in-0.destination=decode-words",
"spring.cloud.stream.bindings.process-out-0.destination=decode-counts",
"spring.cloud.stream.bindings.process-in-0.consumer.useNativeDecoding=false",
"spring.cloud.stream.bindings.process-out-0.producer.useNativeEncoding=false",
"spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId"
+ "=hello-NativeEncodingDecodingEnabledTests-xyz" })
public static class NativeEncodingDecodingDisabledTests
extends KafkaStreamsNativeEncodingDecodingDisabledTests {
@Test
void nativeEncodingDecodingDisabled(EmbeddedKafkaBroker embeddedKafka) {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@Test
public void nativeEncodingDecodingDisabled() {
try (ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.process-in-0.destination=decode-words",
"--spring.cloud.stream.bindings.process-out-0.destination=decode-counts",
"--spring.cloud.stream.bindings.process-in-0.consumer.useNativeDecoding=false",
"--spring.cloud.stream.bindings.process-out-0.producer.useNativeEncoding=false",
"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId=hello-NativeEncodingDecodingEnabledTests-xyz",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("decode-words");
Message<String> msg = MessageBuilder.withPayload("foobar").setHeader("foo", "bar").build();
template.send(msg);
try {
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("decode-words");
Message<String> msg = MessageBuilder.withPayload("foobar").setHeader("foo", "bar").build();
template.send(msg);
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer,
"decode-counts");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer,
"decode-counts");
final Headers headers = cr.headers();
final Iterable<Header> foo = headers.headers("foo");
assertThat(foo.iterator().hasNext()).isTrue();
final Header fooHeader = foo.iterator().next();
assertThat(fooHeader.value()).isEqualTo("bar".getBytes(StandardCharsets.UTF_8));
final Headers headers = cr.headers();
final Iterable<Header> foo = headers.headers("foo");
assertThat(foo.iterator().hasNext()).isTrue();
final Header fooHeader = foo.iterator().next();
assertThat(fooHeader.value()).isEqualTo("bar".getBytes(StandardCharsets.UTF_8));
assertThat(cr.value().equals("Count for foobar : 1")).isTrue();
verify(conversionDelegate).serializeOnOutbound(any(KStream.class));
verify(conversionDelegate).deserializeOnInbound(any(Class.class),
any(KStream.class));
assertThat(cr.value().equals("Count for foobar : 1")).isTrue();
}
finally {
pf.destroy();
}
}
}
@EnableAutoConfiguration
@@ -157,11 +131,10 @@ public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests {
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).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("foo-WordCounts-x"))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5)))
.count(Materialized.as("foo-WordCounts-x"))
.toStream().map((key, value) -> new KeyValue<>(null,
"Count for " + key.key() + " : " + value));
}
}
}

105
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingEnabledTests.java
View File

@@ -31,103 +31,77 @@ import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
/**
* @author Soby Chacko
*/
@RunWith(SpringRunner.class)
@ContextConfiguration
@DirtiesContext
public abstract class KafkaStreamsNativeEncodingDecodingEnabledTests {
@EmbeddedKafka(topics = {"decode-counts", "decode-counts-1"}, partitions = 1)
class KafkaStreamsNativeEncodingDecodingEnabledTests {
/**
* Kafka rule.
*/
@ClassRule
public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true,
"decode-counts", "decode-counts-1");
private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule
.getEmbeddedKafka();
@SpyBean
org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsMessageConversionDelegate conversionDelegate;
private static EmbeddedKafkaBroker embeddedKafka;
private static Consumer<String, String> consumer;
@BeforeClass
@BeforeAll
public static void setUp() {
System.setProperty("spring.cloud.stream.kafka.streams.binder.brokers",
embeddedKafka.getBrokersAsString());
System.setProperty("server.port", "0");
System.setProperty("spring.jmx.enabled", "false");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromEmbeddedTopics(consumer, "decode-counts", "decode-counts-1");
embeddedKafka.consumeFromEmbeddedTopics(consumer, "decode-counts-1");
}
@AfterClass
@AfterAll
public static void tearDown() {
consumer.close();
System.clearProperty("spring.cloud.stream.kafka.streams.binder.brokers");
System.clearProperty("server.port");
System.clearProperty("spring.jmx.enabled");
}
@SpringBootTest(classes = WordCountProcessorApplication.class, properties = {
"spring.cloud.stream.bindings.process-in-0.destination=decode-words-1",
"spring.cloud.stream.bindings.process-out-0.destination=decode-counts-1",
"spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId"
+ "=NativeEncodingDecodingEnabledTests-abc" }, webEnvironment = SpringBootTest.WebEnvironment.NONE)
public static class NativeEncodingDecodingEnabledTests
extends KafkaStreamsNativeEncodingDecodingEnabledTests {
@Test
void nativeEncodingDecodingEnabled(EmbeddedKafkaBroker embeddedKafka) {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
@Test
public void nativeEncodingDecodingEnabled() throws Exception {
try (ConfigurableApplicationContext context = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.process-in-0.destination=decode-words-1",
"--spring.cloud.stream.bindings.process-out-0.destination=decode-counts-1",
"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId=NativeEncodingDecodingEnabledTests-abc",
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("decode-words-1");
template.sendDefault("foobar");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer,
"decode-counts-1");
assertThat(cr.value().equals("Count for foobar : 1")).isTrue();
verify(conversionDelegate, never()).serializeOnOutbound(any(KStream.class));
verify(conversionDelegate, never()).deserializeOnInbound(any(Class.class),
any(KStream.class));
try {
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("decode-words-1");
template.sendDefault("foobar");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer,
"decode-counts-1");
assertThat(cr.value().equals("Count for foobar : 1")).isTrue();
}
finally {
pf.destroy();
}
}
}
@EnableAutoConfiguration
@@ -141,11 +115,10 @@ public abstract class KafkaStreamsNativeEncodingDecodingEnabledTests {
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).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("foo-WordCounts-x"))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5)))
.count(Materialized.as("foo-WordCounts-x"))
.toStream().map((key, value) -> new KeyValue<>(null,
"Count for " + key.key() + " : " + value));
}
}
}

20
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java
View File

@@ -23,8 +23,8 @@ import java.util.function.Consumer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.KStream;
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.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowStore;
@@ -160,15 +160,15 @@ class KafkaStreamsStateStoreIntegrationTests {
@Bean
public Consumer<KStream<Object, Product>> process() {
return input -> input.process(() -> new Processor<Object, Product>() {
return input -> input.process(() -> new Processor<Object, Product, Object, Product>() {
@Override
public void init(ProcessorContext processorContext) {
state = (WindowStore) processorContext.getStateStore("mystate");
public void init(org.apache.kafka.streams.processor.api.ProcessorContext<Object, Product> context) {
state = (WindowStore) context.getStateStore("mystate");
}
@Override
public void process(Object s, Product product) {
public void process(Record<Object, Product> var1) {
processed = true;
}
@@ -202,15 +202,15 @@ class KafkaStreamsStateStoreIntegrationTests {
return (input, input2) -> {
input.process(() -> new Processor<Object, Product>() {
input.process(() -> new Processor<Object, Product, Object, Product>() {
@Override
public void init(ProcessorContext processorContext) {
state = (WindowStore) processorContext.getStateStore("mystate");
public void init(org.apache.kafka.streams.processor.api.ProcessorContext<Object, Product> context) {
state = (WindowStore) context.getStateStore("mystate");
}
@Override
public void process(Object s, Product product) {
public void process(Record<Object, Product> var1) {
processed = true;
}

11
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java
View File

@@ -66,8 +66,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests {
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id",
"false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group-id", false);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
consumerProps);
@@ -81,7 +80,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests {
}
@Test
void kstreamBinderWithPojoInputAndStringOuput() throws Exception {
void kstreamBinderWithPojoInputAndStringOuput(EmbeddedKafkaBroker embeddedKafka) throws Exception {
SpringApplication app = new SpringApplication(ProductCountApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
ConfigurableApplicationContext context = app.run("--server.port=0",
@@ -99,7 +98,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests {
"--spring.cloud.stream.kafka.streams.binder.brokers="
+ embeddedKafka.getBrokersAsString());
try {
receiveAndValidateFoo();
receiveAndValidateFoo(embeddedKafka);
// Assertions on StreamBuilderFactoryBean
StreamsBuilderFactoryBean streamsBuilderFactoryBean = context
.getBean("&stream-builder-process", StreamsBuilderFactoryBean.class);
@@ -113,7 +112,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests {
}
}
private void receiveAndValidateFoo() {
private void receiveAndValidateFoo(EmbeddedKafkaBroker embeddedKafka) throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
senderProps);
@@ -134,7 +133,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests {
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(new JsonSerde<>(Product.class),
new JsonSerde<>(Product.class)))
.windowedBy(TimeWindows.of(Duration.ofMillis(5000)))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000)))
.count(Materialized.as("id-count-store")).toStream()
.map((key, value) -> new KeyValue<>(key.key().id,
"Count for product with ID 123: " + value));

6
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
View File

@@ -162,10 +162,10 @@ class KafkaBinderMetricsTest {
meterRegistry.config().meterFilter(
MeterFilter.denyNameStartsWith("spring.cloud.stream.binder.kafka.offset"));
// Because we have NoopGauge for the offset metric in the meter registry, none of these expectations matter.
// Because we have NoopGauge for the offset metric in the meter registry, none of these expectations matter.
org.mockito.BDDMockito
.given(consumer.committed(ArgumentMatchers.any(TopicPartition.class)))
.willReturn(new OffsetAndMetadata(500));
.given(consumer.committed(ArgumentMatchers.anySet()))
.willReturn(java.util.Map.of(new TopicPartition(TEST_TOPIC, 0), new OffsetAndMetadata(500)));
List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
topicsInUse.put(
TEST_TOPIC,

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

@@ -3553,7 +3553,7 @@ class KafkaBinderTests extends
Consumer consumer = cf.createConsumer();
consumer.subscribe(Collections.singletonList("mixed.0"));
ConsumerRecords records = consumer.poll(10_1000);
ConsumerRecords records = consumer.poll(Duration.ofMillis(10000));
Iterator<ConsumerRecord> iterator = records.iterator();
ConsumerRecord record = iterator.next();
byte[] value = (byte[]) record.value();