Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add diagnostic to KafkaDslTests

Browse Source
This commit is contained in:
Artem Bilan
2024-10-03 12:49:58 -04:00
parent 0998a9502e
commit f47b61d6be
1 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -25,6 +25,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.producer.ProducerConfig;
@@ -98,6 +100,8 @@ import static org.assertj.core.api.Assertions.assertThat;
@DirtiesContext
public class KafkaDslTests {
private static final Log log = LogFactory.getLog(KafkaDslTests.class);
static final String TEST_TOPIC1 = "test-topic1";
static final String TEST_TOPIC2 = "test-topic2";
@@ -179,6 +183,7 @@ public class KafkaDslTests {
for (int i = 0; i < 100; i++) {
Message<?> receive = this.listeningFromKafkaResults1.receive(20000);
assertThat(receive).isNotNull();
log.warn("Received '%s' for index '%d'".formatted(receive, i));
assertThat(receive.getPayload()).isEqualTo("FOO");
MessageHeaders headers = receive.getHeaders();
assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue();
@@ -196,6 +201,7 @@ public class KafkaDslTests {
for (int i = 0; i < 100; i++) {
Message<?> receive = this.listeningFromKafkaResults2.receive(20000);
assertThat(receive).isNotNull();
log.warn("Received '%s' for index '%d'".formatted(receive, i));
assertThat(receive.getPayload()).isEqualTo("FOO");
MessageHeaders headers = receive.getHeaders();
assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue();