From 13f94d266f75d854d6a7c0d85d75022cb7e3272b Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 17 May 2024 17:38:45 -0400 Subject: [PATCH] ReactorKafkaBinderTests race condition issue * There seems to be a race condition in ReactorKafkaBinderTests that causes tests to fail on CI occasionaly. Trying to address this by single dedicated topic per test. --- .../reactorkafka/ReactorKafkaBinderTests.java | 47 +++++++++---------- 1 file changed, 22 insertions(+), 25 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java index 963d5ae96..f42017e6c 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2023 the original author or authors. + * Copyright 2021-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,12 +27,10 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; import org.junit.jupiter.api.Test; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import reactor.kafka.receiver.ReceiverOffset; import reactor.kafka.sender.SenderResult; @@ -68,10 +66,11 @@ import static org.mockito.Mockito.mock; /** * @author Gary Russell + * @author Soby Chacko * @since 4.0 * */ -@EmbeddedKafka(topics = { "testCa", "testCb", "testC1", "testP" }) +@EmbeddedKafka(topics = { "testCa", "testCb", "testC1", "testC-Manual", "testC-AMO", "testP" }) class ReactorKafkaBinderTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @@ -90,7 +89,7 @@ class ReactorKafkaBinderTests { CountDownLatch latch = new CountDownLatch(2); FluxMessageChannel inbound = new FluxMessageChannel(); - Subscriber> sub = new Subscriber>() { + Subscriber> sub = new Subscriber<>() { @Override public void onSubscribe(Subscription s) { @@ -132,16 +131,16 @@ class ReactorKafkaBinderTests { @Test void concurrencyManual() throws Exception { - concurrency(false); + concurrency("testC-Manual", "concurrencyManual-group", false); } @Test void concurrencyAtMostOnce() throws Exception { - concurrency(true); + concurrency("testC-AMO", "concurrencyAtMostOnce-group", true); } @SuppressWarnings({ "rawtypes", "unchecked" }) - void concurrency(boolean atMostOnce) throws Exception { + void concurrency(String topic, String group, boolean atMostOnce) throws Exception { KafkaProperties kafkaProperties = new KafkaProperties(); kafkaProperties.setBootstrapServers( Collections.singletonList(EmbeddedKafkaCondition.getBroker().getBrokersAsString())); @@ -159,7 +158,7 @@ class ReactorKafkaBinderTests { List payloads = Collections.synchronizedList(new ArrayList<>()); FluxMessageChannel inbound = new FluxMessageChannel(); - Subscriber> sub = new Subscriber>() { + Subscriber> sub = new Subscriber<>() { @Override public void onSubscribe(Subscription s) { @@ -192,24 +191,24 @@ class ReactorKafkaBinderTests { KafkaConsumerProperties ext = new KafkaConsumerProperties(); ext.setReactiveAtMostOnce(atMostOnce); ExtendedConsumerProperties props = - new ExtendedConsumerProperties(ext); + new ExtendedConsumerProperties<>(ext); props.setConcurrency(2); - Binding consumer = binder.bindConsumer("testC1", "foo", inbound, props); + Binding consumer = binder.bindConsumer(topic, group, inbound, props); assertThat(subscriptionLatch.await(10, TimeUnit.SECONDS)).isTrue(); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(EmbeddedKafkaCondition.getBroker())); KafkaTemplate kt = new KafkaTemplate<>(pf); - kt.send("testC1", 0, null, "foo").get(10, TimeUnit.SECONDS); - kt.send("testC1", 1, null, "bar").get(10, TimeUnit.SECONDS); - kt.send("testC1", 0, null, "baz").get(10, TimeUnit.SECONDS); - kt.send("testC1", 1, null, "qux").get(10, TimeUnit.SECONDS); + kt.send(topic, 0, null, "foo").get(10, TimeUnit.SECONDS); + kt.send(topic, 1, null, "bar").get(10, TimeUnit.SECONDS); + kt.send(topic, 0, null, "baz").get(10, TimeUnit.SECONDS); + kt.send(topic, 1, null, "qux").get(10, TimeUnit.SECONDS); assertThat(messageLatch1.await(10, TimeUnit.SECONDS)).isTrue(); consumer.stop(); consumer.start(); - kt.send("testC1", 0, null, "fiz").get(10, TimeUnit.SECONDS); - kt.send("testC1", 1, null, "buz").get(10, TimeUnit.SECONDS); + kt.send(topic, 0, null, "fiz").get(10, TimeUnit.SECONDS); + kt.send(topic, 1, null, "buz").get(10, TimeUnit.SECONDS); assertThat(messageLatch2.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(partitions).hasSize(2); consumer.unbind(); @@ -235,11 +234,10 @@ class ReactorKafkaBinderTests { CountDownLatch subscriptionLatch = new CountDownLatch(1); CountDownLatch messageLatch1 = new CountDownLatch(4); - Set partitions = new HashSet<>(); List payloads = Collections.synchronizedList(new ArrayList<>()); FluxMessageChannel inbound = new FluxMessageChannel(); - Subscriber> sub = new Subscriber>() { + Subscriber> sub = new Subscriber<>() { @Override public void onSubscribe(Subscription s) { @@ -250,11 +248,11 @@ class ReactorKafkaBinderTests { @Override public void onNext(Message msg) { ((Message>>) msg).getPayload() - .doOnNext(rec -> { - payloads.add(rec.value()); - messageLatch1.countDown(); - }) - .subscribe(); + .doOnNext(rec -> { + payloads.add(rec.value()); + messageLatch1.countDown(); + }) + .subscribe(); } @Override @@ -339,7 +337,6 @@ class ReactorKafkaBinderTests { props.getExtension().setRecordMetadataChannel("sendResults"); Binding bindProducer = binder.bindProducer("testP", outbound, props); - AtomicReference> sendResult = new AtomicReference<>(); outbound.send(MessageBuilder.withPayload("foo") .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, 1) .build());