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

Fix Poller Config for ambiguity

Browse Source 
Related to https://stackoverflow.com/questions/72023445/spring-cloud-stream-mutually-exclusive-property-issue

The `PollerConfigEnvironmentPostProcessor` has a flaw not checking user-provided properties
and does not check for mutually exclusive before setting default

* Fix `PollerConfigEnvironmentPostProcessor` to check for other mutual poller
properties before setting default for `fixed-delay`
* Same for `max-messages-per-poll`
* Verify that fix is working in the `PollableSourceTests` where poller
options make sense in the Spring Cloud Stream scenario

* Fix `PollerConfigEnvironmentPostProcessor` for redundant check
* Add more tests into `PollableSourceTests` to verify different
poller properties and their exclusive mutuality
This commit is contained in:
Artem Bilan
2022-04-27 13:41:52 -04:00
committed by Soby Chacko
parent 53e6e1a85c
commit 3b32bea133
2 changed files with 94 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/PollerConfigEnvironmentPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2021 the original author or authors.
* Copyright 2021-2022 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.
@@ -56,15 +56,26 @@ public class PollerConfigEnvironmentPostProcessor implements EnvironmentPostProc
}
if (!streamPollerProperties.isEmpty()) {
logger.info("'spring.cloud.stream.poller' properties are deprecated in favor of 'spring.integration.poller' properties.");
logger.info("'spring.cloud.stream.poller' " +
"properties are deprecated in favor of 'spring.integration.poller' properties.");
}
//TODO Must remain after removal of deprecated code above in the future
streamPollerProperties.putIfAbsent(INTEGRATION_PROPERTY_PREFIX + "fixed-delay", "1s");
streamPollerProperties.putIfAbsent(INTEGRATION_PROPERTY_PREFIX + "max-messages-per-poll", "1");
if (!streamPollerProperties.containsKey(INTEGRATION_PROPERTY_PREFIX + "cron") &&
!environment.containsProperty(INTEGRATION_PROPERTY_PREFIX + "cron") &&
!environment.containsProperty(INTEGRATION_PROPERTY_PREFIX + "fixed-rate") &&
!environment.containsProperty(INTEGRATION_PROPERTY_PREFIX + "fixed-delay")) {
environment.getPropertySources()
.addLast(new MapPropertySource("spring.integration.poller", streamPollerProperties));
streamPollerProperties.putIfAbsent(INTEGRATION_PROPERTY_PREFIX + "fixed-delay", "1s");
}
if (!environment.containsProperty(INTEGRATION_PROPERTY_PREFIX + "max-messages-per-poll")) {
streamPollerProperties.putIfAbsent(INTEGRATION_PROPERTY_PREFIX + "max-messages-per-poll", "1");
}
if (!streamPollerProperties.isEmpty()) {
environment.getPropertySources()
.addLast(new MapPropertySource("spring.integration.poller", streamPollerProperties));
}
}
}

83
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/PollableSourceTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2022 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.
@@ -18,42 +18,113 @@ package org.springframework.cloud.stream.function;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
import org.springframework.cloud.stream.binder.PollableMessageSource;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
/**
*
* @author Oleg Zhurakousky
* @author Artem Bilan
*
*/
public class PollableSourceTests {
@Test
void test() {
void testPollableSource() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(PollableAppSampleConfiguration.class))
.web(WebApplicationType.NONE).run(
"--spring.jmx.enabled=false",
"--spring.cloud.stream.pollable-source=blah")) {
"--spring.cloud.stream.pollable-source=blah",
"--spring.integration.poller.cron=*/2 * * * * *",
"--spring.cloud.stream.poller.max-messages-per-poll=4")) {
DefaultPollableMessageSource pollableSource = (DefaultPollableMessageSource) context.getBean(PollableMessageSource.class);
pollableSource.poll(message -> {
assertThat(message.getPayload()).isNotNull();
});
pollableSource.poll(message -> assertThat(message.getPayload()).isNotNull());
PollerMetadata pollerMetadata = context.getBean(PollerMetadata.class);
assertThat(pollerMetadata.getTrigger()).isInstanceOf(CronTrigger.class);
assertThat(TestUtils.getPropertyValue(pollerMetadata.getTrigger(), "expression.expression"))
.isEqualTo("*/2 * * * * *");
assertThat(pollerMetadata.getMaxMessagesPerPoll()).isEqualTo(4);
}
}
@Test
void testPollerDefaultFixedDelay() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(PollableAppSampleConfiguration.class))
.web(WebApplicationType.NONE).run()) {
PollerMetadata pollerMetadata = context.getBean(PollerMetadata.class);
assertThat(pollerMetadata.getTrigger()).isInstanceOf(PeriodicTrigger.class);
assertThat(TestUtils.getPropertyValue(pollerMetadata.getTrigger(), "fixedRate")).isEqualTo(false);
assertThat(TestUtils.getPropertyValue(pollerMetadata.getTrigger(), "period")).isEqualTo(1000L);
assertThat(pollerMetadata.getMaxMessagesPerPoll()).isEqualTo(1);
}
}
@Test
void testPollerProvidedFixedDelayAndMaxMessagesPerPoll() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(PollableAppSampleConfiguration.class))
.web(WebApplicationType.NONE).run(
"--spring.integration.poller.fixed-delay=7s",
"--spring.integration.poller.max-messages-per-poll=13")) {
PollerMetadata pollerMetadata = context.getBean(PollerMetadata.class);
assertThat(pollerMetadata.getTrigger()).isInstanceOf(PeriodicTrigger.class);
assertThat(TestUtils.getPropertyValue(pollerMetadata.getTrigger(), "fixedRate")).isEqualTo(false);
assertThat(TestUtils.getPropertyValue(pollerMetadata.getTrigger(), "period")).isEqualTo(7000L);
assertThat(pollerMetadata.getMaxMessagesPerPoll()).isEqualTo(13);
}
}
@Test
void testNoPollerFixedDelayIfFixedRatePresent() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(PollableAppSampleConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.integration.poller.fixed-rate=200")) {
PollerMetadata pollerMetadata = context.getBean(PollerMetadata.class);
assertThat(pollerMetadata.getTrigger()).isInstanceOf(PeriodicTrigger.class);
assertThat(TestUtils.getPropertyValue(pollerMetadata.getTrigger(), "fixedRate")).isEqualTo(true);
assertThat(TestUtils.getPropertyValue(pollerMetadata.getTrigger(), "period")).isEqualTo(200L);
assertThat(pollerMetadata.getMaxMessagesPerPoll()).isEqualTo(1);
}
}
@Test
void testPollerMutualProperties() {
assertThatExceptionOfType(BeanCreationException.class)
.isThrownBy(() -> new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(PollableAppSampleConfiguration.class))
.web(WebApplicationType.NONE).run(
"--spring.integration.poller.fixed-rate=200",
"--spring.cloud.stream.poller.fixed-delay=300"))
.withRootCauseExactlyInstanceOf(MutuallyExclusiveConfigurationPropertiesException.class)
.withMessageContaining("are mutually exclusive");
}
@EnableAutoConfiguration
@Configuration
public static class PollableAppSampleConfiguration {
}
}