diff --git a/applications/stream-applications-core/stream-applications-test-support/README.adoc b/applications/stream-applications-core/stream-applications-test-support/README.adoc index 79c2f285..d048fee9 100644 --- a/applications/stream-applications-core/stream-applications-test-support/README.adoc +++ b/applications/stream-applications-core/stream-applications-test-support/README.adoc @@ -110,8 +110,8 @@ public class KafkaTimeSourceTests { private OutputMatcher outputMatcher; @Container - static StreamAppContainer timeSource = KafkaConfig.prepackagedContainerFor("time-source", VERSION) - .withLogConsumer(logMatcher); + static StreamAppContainer source = new KafkaStreamAppContainer(StreamAppContainerTestUtils + .imageName(StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY, "time-source-kafka", VERSION)); @Test void test() { await().atMost(DEFAULT_DURATION).until(logMatcher.matches()); @@ -122,7 +122,6 @@ public class KafkaTimeSourceTests { We start an ApplicationContext `KafkaStreamAppContainerTestConfiguration` which starts the `KafkaConfig.kafka` TestContainer in a static initializer. The Time Source emits the time every second. In this case, it's hard to know what the expected output payload is, but it should at least match the date pattern. -This test also uses a `LogMatcher`, which is not strictly necessary, but used here to verify that the app logged the standard start up message - always a good sign. Then we wait for a message on the output topic that matches the pattern. @@ -159,27 +158,35 @@ This realizes the concepts of `source`, `processor`, and `sink` , and similar Sp Here is a test for the canonical `TikTok` stream: ```java -public class RabbitMQTikTokTests extends RabbitMQStreamApplicationIntegrationTestSupport { +@RabbitMQStreamAppTest +public class RabbitTikTokTests { private static LogMatcher logMatcher = LogMatcher.matchesRegex(".*\\d{2}/\\d{2}/\\d{2}\\s+\\d{2}:\\d{2}:\\d{2}") .times(3); @Container - private static final StreamApps streamApp = rabbitMQStreamApps(RabbitMQTikTokTests.class.getSimpleName(), rabbitmq) - .withSourceContainer(prepackagedRabbitMQContainerFor("time-source", VERSION)) - .withSinkContainer(prepackagedRabbitMQContainerFor("log-sink", VERSION) - .withLogConsumer(logMatcher)) - .build(); + private static final StreamApps streamApp = kafkaStreamApps(KafkaTikTokTests.class.getSimpleName(), + KafkaConfig.kafka) + .withSourceContainer( + new RabbitMQStreamAppContainer(StreamAppContainerTestUtils.imageName( + "time-source-rabbit", + VERSION))) + .withSinkContainer( + new RabbitMQStreamAppContainer(StreamAppContainerTestUtils.imageName( + "log-sink-rabbit", + VERSION)).withLogConsumer(logMatcher) + .log()) + .build(); @Test void test() { await().atMost(DEFAULT_DURATION).until(logMatcher.matches()); } +} ``` -Here, the link:src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java[LogMatcher] can be declared statically since it doesn't depend on Spring beans. -This is an extension of TestContainer's `LogConsumer` so it is created before the container starts. Here, we -verify the LogSink logs at least 3 messages that match the pattern. +Here, the link:src/main/java/org/springframework/cloud/stream/app/test/integration/LogMatcher.java[LogMatcher]. +This is an extension of TestContainer's `LogConsumer`. Here, we verify the LogSink logs at least 3 messages that match the pattern. link:src/main/java/org/springframework/cloud/stream/app/test/integration/AppLog.java[AppLog] is also another useful LogConsumer to enable container logging. diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java index 49019028..e2a3e673 100644 --- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/StreamAppContainer.java @@ -94,7 +94,7 @@ public abstract class StreamAppContainer extends GenericContainer>} used to set the version + * dynamically. + */ + Class> versionSupplier() default NullVersionSupplier.class; + + /** + * @return the image name. + */ + String name() default ""; + + /** + * @return the type of Spring Cloud Stream binder used by the application. + */ + Binder binder(); + + /** + * @return the Docker repository name of the image. + */ + String repository() default SPRINGCLOUDSTREAM_REPOSITOTRY; } diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/BaseContainerExtension.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/BaseContainerExtension.java index c632eb4b..fb064e63 100644 --- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/BaseContainerExtension.java +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/BaseContainerExtension.java @@ -16,49 +16,91 @@ package org.springframework.cloud.stream.app.test.integration.junit.jupiter; -import java.util.concurrent.atomic.AtomicInteger; +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; +import java.util.function.Supplier; import org.junit.jupiter.api.extension.ConditionEvaluationResult; import org.junit.jupiter.api.extension.ExecutionCondition; import org.junit.jupiter.api.extension.ExtensionConfigurationException; import org.junit.jupiter.api.extension.ExtensionContext; -import org.junit.platform.commons.util.AnnotationUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; +import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils; +import org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer; +import org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer; +import org.springframework.core.annotation.AnnotatedElementUtils; /** - * A junit Jupiter extension used to discover {@link StreamAppContainer}s annotated with {code @BaseContainer}. + * A junit Jupiter extension used to discover {@link StreamAppContainer}s annotated with + * {@link BaseContainer}. * @author David Turanski */ public class BaseContainerExtension implements ExecutionCondition { private static StreamAppContainer baseContainer; + private static Logger logger = LoggerFactory.getLogger(BaseContainerExtension.class); + public static StreamAppContainer containerInstance() { return baseContainer; } @Override public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext extensionContext) { - AtomicInteger count = new AtomicInteger(); - AnnotationUtils.findAnnotatedFields(extensionContext.getRequiredTestClass(), BaseContainer.class, field -> { - try { - StreamAppContainer base = (StreamAppContainer) field.get(null); - if (base == null) { - throw new ExtensionConfigurationException("@BaseContainer is not initialized"); - } - count.getAndIncrement(); - baseContainer = base; - } - catch (Exception e) { - throw new ExtensionConfigurationException(e.getMessage(), e); - } - return true; - }); - if (count.get() != 1) { - throw new ExtensionConfigurationException( - "Expecting exactly one @BaseContainer instance, found " + count.get()); + BaseContainer annotation = AnnotatedElementUtils.getMergedAnnotation(extensionContext.getRequiredTestClass(), + BaseContainer.class); + if (annotation == null) { + throw new ExtensionConfigurationException("@BaseContainer is required for this extension"); } - return ConditionEvaluationResult.enabled("@BaseContainer found"); + + String version = getVersion(annotation); + + switch (annotation.binder()) { + case Kafka: + baseContainer = new KafkaStreamAppContainer( + StreamAppContainerTestUtils.imageName(annotation.repository(), annotation.name(), version)); + break; + case RabbitMQ: + baseContainer = new RabbitMQStreamAppContainer( + StreamAppContainerTestUtils.imageName(annotation.repository(), annotation.name(), version)); + break; + default: + throw new ExtensionConfigurationException( + "the binder type " + annotation.binder().name() + " is not supported"); + } + logger.debug("StreamAppContainer created using base container image " + baseContainer.getDockerImageName()); + + return ConditionEvaluationResult + .enabled("StreamAppContainer created using base container image " + baseContainer.getDockerImageName()); + } + + private String getVersion(BaseContainer annotation) { + String version = null; + if (annotation.version().isEmpty() && annotation.versionSupplier().equals(NullVersionSupplier.class)) { + throw new ExtensionConfigurationException( + "either 'version' or 'versionSupplier' must be set in @BaseContainer"); + } + if (!annotation.version().isEmpty() && !annotation.versionSupplier().equals(NullVersionSupplier.class)) { + throw new ExtensionConfigurationException( + "only one of 'version' or 'versionSupplier' must be set in @BaseContainer"); + } + if (annotation.version().isEmpty()) { + try { + Constructor> constructor = annotation.versionSupplier().getConstructor(); + version = constructor.newInstance().get(); + } + catch (NoSuchMethodException | InstantiationException | IllegalAccessException + | InvocationTargetException e) { + throw new ExtensionConfigurationException( + "No accessible default constructor found for " + annotation.versionSupplier().getName()); + } + } + else { + version = annotation.version(); + } + return version; } } diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/Binder.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/Binder.java new file mode 100644 index 00000000..a7dc2d3d --- /dev/null +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/Binder.java @@ -0,0 +1,33 @@ +/* + * Copyright 2020-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.test.integration.junit.jupiter; + +/** + * Supported Spring Cloud Stream binder types. + * @author David Turanski + */ +public enum Binder { + /** + * Kafka Binder. + */ + Kafka, + + /** + * RabbitMQ Binder. + */ + RabbitMQ, +} diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/KafkaBaseContainer.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/KafkaBaseContainer.java new file mode 100644 index 00000000..577f5642 --- /dev/null +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/KafkaBaseContainer.java @@ -0,0 +1,50 @@ +/* + * Copyright 2020-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.test.integration.junit.jupiter; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import java.util.function.Supplier; + +import org.springframework.core.annotation.AliasFor; + +import static org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY; + +/** + * Annotation For configuring a {@link org.springframework.cloud.stream.app.test.integration.kafka.KafkaStreamAppContainer}. + * @author David Turanski + * @see org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainer + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +@BaseContainer(binder = Binder.Kafka) +public @interface KafkaBaseContainer { + + @AliasFor(annotation = BaseContainer.class) + String version() default ""; + + @AliasFor(annotation = BaseContainer.class) + Class> versionSupplier() default NullVersionSupplier.class; + + @AliasFor(annotation = BaseContainer.class) + String name() default ""; + + @AliasFor(annotation = BaseContainer.class) + String repository() default SPRINGCLOUDSTREAM_REPOSITOTRY; +} diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/NullVersionSupplier.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/NullVersionSupplier.java new file mode 100644 index 00000000..37c60c08 --- /dev/null +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/NullVersionSupplier.java @@ -0,0 +1,31 @@ +/* + * Copyright 2020-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.test.integration.junit.jupiter; + +import java.util.function.Supplier; + +/** + * A placeHolder for {@link org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainer} + * indicating no "versionSupplier" has been set. + * @author David Turanski + */ +class NullVersionSupplier implements Supplier { + @Override + public String get() { + return null; + } +} diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/RabbitMQBaseContainer.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/RabbitMQBaseContainer.java new file mode 100644 index 00000000..b4371fce --- /dev/null +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/junit/jupiter/RabbitMQBaseContainer.java @@ -0,0 +1,50 @@ +/* + * Copyright 2020-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.test.integration.junit.jupiter; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import java.util.function.Supplier; + +import org.springframework.core.annotation.AliasFor; + +import static org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils.SPRINGCLOUDSTREAM_REPOSITOTRY; + +/** + * Annotation For configuring a + * {@link org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer}. + * @author David Turanski + * @see org.springframework.cloud.stream.app.test.integration.junit.jupiter.BaseContainer + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +@BaseContainer(binder = Binder.RabbitMQ) +public @interface RabbitMQBaseContainer { + @AliasFor(annotation = BaseContainer.class) + String version() default ""; + + @AliasFor(annotation = BaseContainer.class) + Class> versionSupplier() default NullVersionSupplier.class; + + @AliasFor(annotation = BaseContainer.class) + String name() default ""; + + @AliasFor(annotation = BaseContainer.class) + String repository() default SPRINGCLOUDSTREAM_REPOSITOTRY; +} diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java index aaeff90a..d29963d3 100644 --- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaConfig.java @@ -20,24 +20,14 @@ import org.testcontainers.containers.KafkaContainer; import org.testcontainers.containers.Network; import org.testcontainers.utility.DockerImageName; -import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; -import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils; - /** - * Initializes and starts a Kafka TestContainer and provides associated utilities for configuring a {@link KafkaStreamAppContainer}. + * Initializes and starts a {@link KafkaContainer}. * @author David Turanski */ public abstract class KafkaConfig { - final static String BINDER = "kafka"; final static Network network = Network.SHARED; - public static StreamAppContainer prepackagedContainerFor(String appName, String version) { - return new KafkaStreamAppContainer( - StreamAppContainerTestUtils.prePackagedStreamAppImageName(appName, BINDER, version), - kafka); - } - /** * The KafkaContainer. */ diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java index 2fa00134..6c78e276 100644 --- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/kafka/KafkaStreamAppContainer.java @@ -16,8 +16,6 @@ package org.springframework.cloud.stream.app.test.integration.kafka; -import org.testcontainers.containers.GenericContainer; - import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; /** @@ -30,10 +28,9 @@ public class KafkaStreamAppContainer extends StreamAppContainer { /** * @param imageName the image name. - * @param kafka a running kafka TestContainer instance. */ - public KafkaStreamAppContainer(String imageName, GenericContainer kafka) { - super(imageName, kafka); + public KafkaStreamAppContainer(String imageName) { + super(imageName, KafkaConfig.kafka); } @Override diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java index 03c7867f..6f9f2ae2 100644 --- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQConfig.java @@ -20,13 +20,8 @@ import org.testcontainers.containers.Network; import org.testcontainers.containers.RabbitMQContainer; import org.testcontainers.utility.DockerImageName; -import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; -import org.springframework.cloud.stream.app.test.integration.StreamAppContainerTestUtils; - /** - * Initializes and starts a RabbitMQ TestContainer and provides associated utilities for - * configuring a - * {@link org.springframework.cloud.stream.app.test.integration.rabbitmq.RabbitMQStreamAppContainer}. + * Initializes and starts a {@link RabbitMQContainer}. * @author David Turanski */ public abstract class RabbitMQConfig { @@ -35,8 +30,6 @@ public abstract class RabbitMQConfig { */ public static RabbitMQContainer rabbitmq; - final static String BINDER = "rabbit"; - final static Network network = Network.SHARED; static { @@ -45,11 +38,4 @@ public abstract class RabbitMQConfig { .withExposedPorts(5672, 15672); rabbitmq.start(); } - - public static StreamAppContainer prepackagedContainerFor(String appName, String version) { - return new RabbitMQStreamAppContainer( - StreamAppContainerTestUtils.prePackagedStreamAppImageName(appName, BINDER, version), - rabbitmq); - } - } diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java index 34e3db0d..d6a0ac4f 100644 --- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainer.java @@ -16,8 +16,6 @@ package org.springframework.cloud.stream.app.test.integration.rabbitmq; -import org.testcontainers.containers.GenericContainer; - import org.springframework.cloud.stream.app.test.integration.StreamAppContainer; /** @@ -30,10 +28,10 @@ public class RabbitMQStreamAppContainer extends StreamAppContainer { /** * @param imageName the image name. - * @param rabbitmq a running rabbitMQ TestContainer instance. */ - public RabbitMQStreamAppContainer(String imageName, GenericContainer rabbitmq) { - super(imageName, rabbitmq); + public RabbitMQStreamAppContainer(String imageName) { + + super(imageName, RabbitMQConfig.rabbitmq); } @Override diff --git a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java index 29062f9b..4c4e6d26 100644 --- a/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java +++ b/applications/stream-applications-core/stream-applications-test-support/src/main/java/org/springframework/cloud/stream/app/test/integration/rabbitmq/RabbitMQStreamAppContainerTestConfiguration.java @@ -292,7 +292,7 @@ public abstract class RabbitMQStreamAppContainerTestConfiguration { @Override public

void send(String topic, P payload) { - await().atMost(Duration.ofSeconds(30)).pollDelay(Duration.ofSeconds(1)).pollInterval(Duration.ofSeconds(1)) + await().atMost(Duration.ofSeconds(30)).pollInterval(Duration.ofSeconds(1)) .until(topicExistsAndIsBound(topic)); rabbitTemplate.convertAndSend(topic, routingKey, payload); }