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

Address compile issues in Kafka binder tests

Browse Source
This commit is contained in:
Soby Chacko
2022-11-03 14:46:33 -04:00
parent 81ddec3ce3
commit 330610fde9
2 changed files with 5 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.kafka.streams.function;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -169,7 +170,7 @@ public class KafkaStreamsComponentBeansTests {
template.sendDefault("foobar");
template.setDefaultTopic("testBiFunctionComponent-in-1");
template.sendDefault("foobar");
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer2, 10_000, 2);
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer2, Duration.ofSeconds(10), 2);
assertThat(records.count()).isEqualTo(2);
records.forEach(stringStringConsumerRecord -> assertThat(stringStringConsumerRecord.value().contains("foobar")).isTrue());
}
@@ -259,7 +260,7 @@ public class KafkaStreamsComponentBeansTests {
template.sendDefault("foobar");
template.setDefaultTopic("testCurriedFunctionWithFunctionTerminal-in-2");
template.sendDefault("foobar");
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer3, 10_000, 3);
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer3, Duration.ofSeconds(10), 3);
assertThat(records.count()).isEqualTo(3);
records.forEach(stringStringConsumerRecord -> assertThat(stringStringConsumerRecord.value().contains("foobar")).isTrue());
}

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.kafka.streams.integration;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -164,7 +165,7 @@ public class KafkaStreamsBinderHealthIndicatorTests {
latch.await(5, TimeUnit.SECONDS);
embeddedKafka.consumeFromEmbeddedTopics(consumer, topics);
KafkaTestUtils.getRecords(consumer, 1000);
KafkaTestUtils.getRecords(consumer, Duration.ofSeconds(1));
TimeUnit.SECONDS.sleep(5);
checkHealth(context, expected);