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

Fix checkstyles

Browse Source
This commit is contained in:
Oleg Zhurakousky
2024-10-21 17:57:07 +02:00
parent fa9a80d803
commit 9888a08dff
24 changed files with 379 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java
View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.stream.binder.kafka.properties;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import javax.security.auth.login.AppConfigurationEntry;
@@ -55,7 +56,7 @@ public class JaasLoginModuleConfiguration {
public void setControlFlag(String controlFlag) {
Assert.notNull(controlFlag, "cannot be null");
this.controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag
.valueOf(controlFlag.toUpperCase());
.valueOf(controlFlag.toUpperCase(Locale.ROOT));
}
public Map<String, String> getOptions() {

3
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java
View File

@@ -20,6 +20,7 @@ import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
@@ -193,7 +194,7 @@ class ReactorKafkaBinderIntegrationTests {
@Bean
Function<Flux<ReceiverRecord<byte[], byte[]>>, Flux<String>> lowercase() {
return s -> s.map(rec -> new String(rec.value()).toLowerCase());
return s -> s.map(rec -> new String(rec.value()).toLowerCase(Locale.ROOT));
}
@Bean

161
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java Normal file
View File

@@ -0,0 +1,161 @@
/*
* Copyright 2022-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.
* 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.binder.reactorkafka;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.Locale;
import java.util.function.Function;
import java.util.stream.Collectors;
import brave.handler.SpanHandler;
import brave.test.TestSpanHandler;
import io.micrometer.observation.Observation;
import io.micrometer.observation.ObservationRegistry;
import io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor;
import io.micrometer.tracing.brave.bridge.BraveFinishedSpan;
import io.micrometer.tracing.test.simple.SpansAssert;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.kafka.receiver.ReceiverRecord;
import reactor.kafka.receiver.observation.KafkaReceiverObservation;
import reactor.kafka.receiver.observation.KafkaRecordReceiverContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.autoconfigure.actuate.observability.AutoConfigureObservability;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
/**
* @author Artem Bilan
* @author Soby Chacko
* @since 4.2.0
*/
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
"spring.kafka.consumer.metadata.max.age.ms=1000",
"spring.cloud.function.definition=receive",
"spring.cloud.stream.function.reactive.uppercase=true",
"spring.cloud.stream.bindings.receive-in-0.group=rkbot-in-group",
"spring.cloud.stream.bindings.receive-in-0.destination=rkbot-in-topic",
"spring.cloud.stream.bindings.receive-out-0.destination=rkbot-out-topic",
"spring.cloud.stream.kafka.binder.enable-observation=true",
"spring.cloud.stream.kafka.binder.brokers=${spring.kafka.bootstrap-servers}",
"management.tracing.sampling.probability=1",
"spring.cloud.stream.kafka.bindings.receive-in-0.consumer.converterBeanName=fullRR"
})
@DirtiesContext
@AutoConfigureObservability
@EmbeddedKafka(topics = { "rkbot-out-topic" })
public class ReactorKafkaBinderObservationTests {
private static final TestSpanHandler SPANS = new TestSpanHandler();
@Autowired
StreamBridge streamBridge;
@Autowired
ObservationRegistry observationRegistry;
@Autowired
TestConfiguration testConfiguration;
@Autowired
private EmbeddedKafkaBroker embeddedKafka;
@Test
void endToEndReactorKafkaBinder1() {
streamBridge.send("rkbot-in-topic", MessageBuilder.withPayload("data")
.build());
await().timeout(Duration.ofSeconds(10)).untilAsserted(() -> assertThat(SPANS.spans()).hasSize(3));
SpansAssert.assertThat(SPANS.spans().stream().map(BraveFinishedSpan::fromBrave).collect(Collectors.toList()))
.haveSameTraceId();
}
@SpringBootConfiguration
@EnableAutoConfiguration(exclude = org.springframework.cloud.function.observability.ObservationAutoConfiguration.class)
public static class TestConfiguration {
@Bean
SpanHandler testSpanHandler() {
return SPANS;
}
@Bean
RecordMessageConverter fullRR() {
return new RecordMessageConverter() {
private final RecordMessageConverter converter = new MessagingMessageConverter();
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
org.apache.kafka.clients.consumer.Consumer<?, ?> consumer, Type payloadType) {
return MessageBuilder.withPayload(record).build();
}
@Override
public ProducerRecord<?, ?> fromMessage(Message<?> message, String defaultTopic) {
return this.converter.fromMessage(message, defaultTopic);
}
};
}
@Bean
Function<Flux<ReceiverRecord<byte[], byte[]>>, Flux<Message<String>>> receive(ObservationRegistry observationRegistry) {
return s -> s
.flatMap(record -> {
Observation receiverObservation =
KafkaReceiverObservation.RECEIVER_OBSERVATION.start(null,
KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention.INSTANCE,
() ->
new KafkaRecordReceiverContext(
record, "user.receiver", "localhost:9092"),
observationRegistry);
return Mono.deferContextual(contextView -> Mono.just(record)
.map(rec -> new String(rec.value()).toLowerCase(Locale.ROOT))
.map(rec -> MessageBuilder.withPayload(rec).setHeader(IntegrationMessageHeaderAccessor.REACTOR_CONTEXT, contextView).build()))
.doOnTerminate(receiverObservation::stop)
.doOnError(receiverObservation::error)
.contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, receiverObservation));
});
}
}
}

3
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -370,7 +371,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
if (!concurrencyExplicitlyProvided[0]) {
concurrencyExplicitlyProvided[0] = name.getLastElement(ConfigurationPropertyName.Form.UNIFORM).equals("concurrency") &&
// name is normalized to contain only uniform elements and thus safe to call toLowerCase here.
ConfigurationPropertyName.of("spring.cloud.stream.bindings." + inboundName.toLowerCase() + ".consumer")
ConfigurationPropertyName.of("spring.cloud.stream.bindings." + inboundName.toLowerCase(Locale.ROOT) + ".consumer")
.isAncestorOf(name);
}
return result;

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.function;
import java.time.Duration;
import java.util.Arrays;
import java.util.Date;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
@@ -190,7 +191,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests {
return input -> {
final Map<String, KStream<Object, WordCount>> stringKStreamMap = input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.groupBy((key, value) -> value)
.windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
.count(Materialized.as("WordCounts-branch"))

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

@@ -21,6 +21,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
@@ -397,7 +398,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
Function<KStream<Object, String>, KStream<String, WordCount>> process() {
return input -> input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofMillis(5000)))
@@ -439,7 +440,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
Function<KStream<Object, String>, KStream<?, WordCount>> process() {
return input -> input
.flatMapValues(
value -> Arrays.asList(value.toLowerCase().split("\\W+")))
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5))).count(Materialized.as("foobar-WordCounts"))

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

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.integration;
import java.time.Duration;
import java.util.Arrays;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
@@ -124,7 +125,7 @@ class DlqDestinationResolverTests {
return input -> input
.flatMapValues(
value -> Arrays.asList(value.toLowerCase().split("\\W+")))
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x"))

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.integration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
@@ -149,7 +150,7 @@ class KafkaStreamsBinderMultipleInputTopicsTest {
return input -> input
.flatMapValues(
value -> Arrays.asList(value.toLowerCase().split("\\W+")))
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.count(Materialized.as("WordCounts-tKWCWSIAP0")).toStream()

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.integration;
import java.time.Duration;
import java.util.Arrays;
import java.util.Date;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.function.Function;
@@ -173,7 +174,7 @@ class KafkaStreamsBinderTombstoneTests {
public Function<KStream<Object, String>, KStream<String, WordCount>> process() {
return input -> input
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofMillis(5000)))

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.integration;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
@@ -153,7 +154,7 @@ public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests {
return input -> input
.flatMapValues(
value -> Arrays.asList(value.toLowerCase().split("\\W+")))
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x"))

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

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.integration;
import java.time.Duration;
import java.util.Arrays;
import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
@@ -137,7 +138,7 @@ public abstract class KafkaStreamsNativeEncodingDecodingEnabledTests {
return input -> input
.flatMapValues(
value -> Arrays.asList(value.toLowerCase().split("\\W+")))
value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+")))
.map((key, value) -> new KeyValue<>(value, value))
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x"))

3
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -26,6 +26,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -1517,7 +1518,7 @@ public class KafkaMessageChannelBinder extends
if (producerProperties.isDynamicPartitionUpdatesEnabled() &&
producerProperties.getPartitionKeyExpression() != null &&
!(producerProperties.getPartitionKeyExpression().getExpressionString()
.toLowerCase().contains("payload"))) {
.toLowerCase(Locale.ROOT).contains("payload"))) {
kafkaPartitionHandler =
new PartitionHandler(ExpressionUtils.createStandardEvaluationContext(beanFactory),
producerProperties, beanFactory);

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.kafka.integration;
import java.util.Locale;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
@@ -86,7 +87,7 @@ class ProducerOnlyTransactionTests {
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.name().toLowerCase());
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.name().toLowerCase(Locale.ROOT));
Consumer<?, ?> consumer = new KafkaConsumer<>(props);
embeddedKafkaBrokera.consumeFromAllEmbeddedTopics(consumer);
ConsumerRecord<?, ?> record = KafkaTestUtils.getSingleRecord(consumer, "output");

3
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.integration2;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -128,7 +129,7 @@ class ConsumerProducerTransactionTests {
if (in.equals("two")) {
throw new RuntimeException("fail");
}
return in.toUpperCase();
return in.toUpperCase(Locale.ROOT);
};
}

9
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
@@ -116,7 +117,7 @@ class PollableConsumerTests {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
return MessageBuilder
.withPayload(((String) message.getPayload()).toUpperCase())
.withPayload(((String) message.getPayload()).toUpperCase(Locale.ROOT))
.copyHeaders(message.getHeaders()).build();
}
@@ -301,7 +302,7 @@ class PollableConsumerTests {
public Message<?> preSend(Message<?> message, MessageChannel channel) {
return MessageBuilder
.withPayload(
new String((byte[]) message.getPayload()).toUpperCase())
new String((byte[]) message.getPayload()).toUpperCase(Locale.ROOT))
.copyHeaders(message.getHeaders()).build();
}
@@ -328,7 +329,7 @@ class PollableConsumerTests {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
return MessageBuilder
.withPayload(((String) message.getPayload()).toUpperCase())
.withPayload(((String) message.getPayload()).toUpperCase(Locale.ROOT))
.copyHeaders(message.getHeaders()).build();
}
@@ -379,7 +380,7 @@ class PollableConsumerTests {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
return MessageBuilder
.withPayload(((String) message.getPayload()).toUpperCase())
.withPayload(((String) message.getPayload()).toUpperCase(Locale.ROOT))
.copyHeaders(message.getHeaders()).build();
}

5
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/FunctionPostProcessingTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.function;
import java.util.Locale;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -241,7 +242,7 @@ class FunctionPostProcessingTests {
public Function<String, String> uppercase() {
return new PostProcessingFunction<String, String>() {
public String apply(String input) {
return input.toUpperCase();
return input.toUpperCase(Locale.ROOT);
}
public void postProcess(Message<String> result) {
@@ -260,7 +261,7 @@ class FunctionPostProcessingTests {
if (input.equals("error")) {
throw new RuntimeException("intentional");
}
return input.toUpperCase();
return input.toUpperCase(Locale.ROOT);
}
@Override

3
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.function;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Locale;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -94,7 +95,7 @@ class GreenfieldFunctionEnableBindingTests {
@Bean
public Function<String, String> toUpperCase() {
return String::toUpperCase;
return v -> v.toUpperCase(Locale.ROOT);
}
}

162
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/HeaderTests.java Normal file
View File

@@ -0,0 +1,162 @@
/*
* Copyright 2024-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.
* 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.function;
import java.util.Locale;
import java.util.function.Function;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.function.context.message.MessageUtils;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Omer Celik
*/
public class HeaderTests {
@BeforeAll
public static void before() {
System.clearProperty("spring.cloud.function.definition");
}
@Test
void checkWithEmptyPojo() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
StreamBridge streamBridge = context.getBean(StreamBridge.class);
Message<EmptyPojo> message = MessageBuilder.withPayload(new EmptyPojo()).build();
streamBridge.send("emptyConfigurationDestination", message);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> messageReceived = outputDestination.receive(1000, "emptyConfigurationDestination");
MessageHeaders headers = messageReceived.getHeaders();
assertThat(headers).isNotNull();
assertThat(headers.get(MessageUtils.TARGET_PROTOCOL)).isEqualTo("kafka");
assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json");
}
}
@Test
void checkIfHeaderProvidedInData() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) {
StreamBridge streamBridge = context.getBean(StreamBridge.class);
String jsonPayload = "{\"name\":\"Omer\"}";
streamBridge.send("myBinding-out-0",
MessageBuilder.withPayload(jsonPayload.getBytes())
.setHeader("anyHeader", "anyValue")
.build(),
MimeTypeUtils.APPLICATION_JSON);
OutputDestination output = context.getBean(OutputDestination.class);
Message<byte[]> result = output.receive(1000, "myBinding-out-0");
MessageHeaders headers = result.getHeaders();
assertThat(headers).isNotNull();
assertThat(headers.get(MessageUtils.TARGET_PROTOCOL)).isEqualTo("kafka");
assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json");
assertThat(headers.get("anyHeader")).isEqualTo("anyValue");
}
}
@Test
void checkGenericMessageSent() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionUpperCaseConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=uppercase")) {
String jsonPayload = "{\"surname\":\"Celik\"}";
InputDestination input = context.getBean(InputDestination.class);
input.send(new GenericMessage<>(jsonPayload.getBytes()), "uppercase-in-0");
OutputDestination output = context.getBean(OutputDestination.class);
Message<byte[]> result = output.receive(1000, "uppercase-out-0");
MessageHeaders headers = result.getHeaders();
assertThat(headers).isNotNull();
assertThat(headers.get(MessageUtils.TARGET_PROTOCOL)).isEqualTo("kafka");
assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json");
}
}
@Test
void checkMessageWrappedFunctionalConsumer() {
System.clearProperty("spring.cloud.function.definition");
ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionMessageConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=uppercase"
);
InputDestination source = context.getBean(InputDestination.class);
source.send(new GenericMessage<>("Omer Celik".getBytes()), "uppercase-in-0");
OutputDestination target = context.getBean(OutputDestination.class);
Message<byte[]> message = target.receive(5, "uppercase-out-0");
MessageHeaders headers = message.getHeaders();
assertThat(headers).isNotNull();
assertThat(headers).isNotNull();
assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json");
assertThat(headers.get(MessageUtils.TARGET_PROTOCOL)).isEqualTo("kafka");
}
@EnableAutoConfiguration
public static class EmptyConfiguration {
}
@EnableAutoConfiguration
public static class FunctionMessageConfiguration {
@Bean
public Function<Message<String>, Message<String>> uppercase() {
return msg -> MessageBuilder.withPayload(msg.getPayload().toUpperCase(Locale.ROOT)).build();
}
}
@EnableAutoConfiguration
@Configuration
public static class FunctionUpperCaseConfiguration {
@Bean
public Function<String, String> uppercase() {
return String::toUpperCase;
}
}
public static class EmptyPojo {
}
}

7
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -24,6 +24,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@@ -245,7 +246,7 @@ class ImplicitFunctionBindingTests {
// good, we expected it
}
Function<byte[], String> function = v -> new String(v).toUpperCase();
Function<byte[], String> function = v -> new String(v).toUpperCase(Locale.ROOT);
FunctionBindingTestUtils.bind(context, function);
input.send(new GenericMessage<byte[]>("hello".getBytes()));
@@ -1434,7 +1435,7 @@ class ImplicitFunctionBindingTests {
public Function<Flux<Message<Person>>, Flux<Message<Person>>> reactivePojoMessage() {
return flux -> flux.map(message -> {
Person p = message.getPayload();
p.setName(p.getName().toUpperCase());
p.setName(p.getName().toUpperCase(Locale.ROOT));
return MessageBuilder.withPayload(p).copyHeaders(message.getHeaders()).build();
});
}
@@ -1636,7 +1637,7 @@ class ImplicitFunctionBindingTests {
@Bean
public Function<String, String> uppercase() {
return v -> v.toUpperCase();
return v -> v.toUpperCase(Locale.ROOT);
}
@Bean

7
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.function;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Locale;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -296,7 +297,7 @@ class MultipleInputOutputFunctionTests {
@Bean
public Function<String, String> uppercase() {
return value -> value.toUpperCase();
return value -> value.toUpperCase(Locale.ROOT);
}
@Bean
@@ -378,8 +379,8 @@ class MultipleInputOutputFunctionTests {
@Bean
public Function<Tuple2<Flux<Person>, Flux<Employee>>, Flux<String>> multiInputSingleOutput() {
return tuple -> {
Flux<String> stringStream = tuple.getT1().map(p -> p.getName().toUpperCase());
Flux<String> intStream = tuple.getT2().map(p -> p.getName().toUpperCase());
Flux<String> stringStream = tuple.getT1().map(p -> p.getName().toUpperCase(Locale.ROOT));
Flux<String> intStream = tuple.getT2().map(p -> p.getName().toUpperCase(Locale.ROOT));
return Flux.merge(stringStream, intStream);
};
}

5
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/RoutingFunctionTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.function;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Locale;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -376,7 +377,6 @@ class RoutingFunctionTests {
@Bean
public Function<Message<String>, Message<String>> enrich() {
return x -> {
System.out.println("===> enrich");
return MessageBuilder.withPayload(x.getPayload()).setHeader("spring.cloud.function.definition", "uppercase").build();
};
}
@@ -384,8 +384,7 @@ class RoutingFunctionTests {
@Bean
public Function<String, String> uppercase() {
return x -> {
System.out.println("===> uppercase");
return x.toUpperCase();
return x.toUpperCase(Locale.ROOT);
};
}

5
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.function;
import java.lang.reflect.Field;
import java.net.URI;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -819,7 +820,7 @@ class StreamBridgeTests {
public static class DynamicProducerDestinationConfig {
@Bean
public Function<Message<String>, Message<String>> uppercase() {
return msg -> MessageBuilder.withPayload(msg.getPayload().toUpperCase())
return msg -> MessageBuilder.withPayload(msg.getPayload().toUpperCase(Locale.ROOT))
.setHeader("spring.cloud.stream.sendto.destination", "dynamicTopic").build();
}
}
@@ -1003,7 +1004,7 @@ class StreamBridgeTests {
public IntegrationFlow transform(StreamBridge bridge) {
return IntegrationFlow.from("foo").transform(v -> {
String s = new String((byte[]) v);
return s.toUpperCase();
return s.toUpperCase(Locale.ROOT);
})
.handle(v -> bridge.send("output", v))
.get();

5
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -26,6 +26,7 @@ import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
@@ -212,7 +213,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
}
private <T> String getKafkaStreamsBinderSimpleName(Class<? extends T> bindingTargetType) {
return bindingTargetType.getSimpleName().toLowerCase();
return bindingTargetType.getSimpleName().toLowerCase(Locale.ROOT);
}
private <T> boolean isKafkaStreamsType(Class<? extends T> bindingTargetType) {
@@ -225,7 +226,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
if (!MessageChannel.class.isAssignableFrom(bindingTargetType)
&& !PollableMessageSource.class.isAssignableFrom(bindingTargetType)) {
String bindingTargetTypeName = StringUtils.hasText(name) ? name : bindingTargetType.getSimpleName().toLowerCase();
String bindingTargetTypeName = StringUtils.hasText(name) ? name : bindingTargetType.getSimpleName().toLowerCase(Locale.ROOT);
Binder<T, ConsumerProperties, ProducerProperties> binderInstance = getBinderInstance(bindingTargetTypeName);
return binderInstance;
}

5
schema-registry/spring-cloud-stream-schema-registry-server/src/test/java/org/springframework/cloud/stream/schema/registry/server/SchemaRegistryServerAvroTests.java
View File

@@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
@@ -81,12 +82,12 @@ class SchemaRegistryServerAvroTests {
.parse(resourceToString("classpath:/avro_user_definition_schema_v2.json"));
private static final String AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT = AVRO_USER_AVRO_SCHEMA_V1.getName()
.toLowerCase();
.toLowerCase(Locale.ROOT);
private static final String AVRO_USER_SCHEMA_QUALIFED_NAME_STRATEGY_SUBJECT = AVRO_USER_AVRO_SCHEMA_V1
.getFullName()
.toLowerCase();
.toLowerCase(Locale.ROOT);
private static final Schema AVRO_USER_REGISTRY_SCHEMA_V1 = toSchema(
AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT,