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

Default properties testing.

Browse Source 
Remove camelCase in property names
This commit is contained in:
Soby Chacko
2018-10-08 12:29:20 -04:00
parent 1d5fcca522
commit fe7bdd245d
3 changed files with 24 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java
View File

@@ -70,10 +70,10 @@ public class StreamToGlobalKTableJoinIntegrationTests {
interface CustomGlobalKTableProcessor extends KafkaStreamsProcessor {
@Input("inputX")
@Input("input-x")
GlobalKTable<?, ?> inputX();
@Input("inputY")
@Input("input-y")
GlobalKTable<?, ?> inputY();
}
@@ -85,8 +85,8 @@ public class StreamToGlobalKTableJoinIntegrationTests {
@StreamListener
@SendTo("output")
public KStream<Long, EnrichedOrder> process(@Input("input") KStream<Long, Order> ordersStream,
@Input("inputX") GlobalKTable<Long, Customer> customers,
@Input("inputY") GlobalKTable<Long, Product> products) {
@Input("input-x") GlobalKTable<Long, Customer> customers,
@Input("input-y") GlobalKTable<Long, Product> products) {
KStream<Long, CustomerOrder> customerOrdersStream = ordersStream.join(customers,
(orderId, order) -> order.getCustomerId(),
@@ -112,19 +112,19 @@ public class StreamToGlobalKTableJoinIntegrationTests {
try (ConfigurableApplicationContext ignored = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.destination=orders",
"--spring.cloud.stream.bindings.inputX.destination=customers",
"--spring.cloud.stream.bindings.inputY.destination=products",
"--spring.cloud.stream.bindings.input-x.destination=customers",
"--spring.cloud.stream.bindings.input-y.destination=products",
"--spring.cloud.stream.bindings.output.destination=enriched-order",
"--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.inputX.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.inputY.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.input-x.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.input-y.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$OrderSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$CustomerSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputY.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.inputY.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$ProductSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$CustomerSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$ProductSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$LongSerde",
"--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.springframework.cloud.stream.binder.kafka.streams.integration.StreamToGlobalKTableJoinIntegrationTests$EnrichedOrderSerde",
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",

8
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java
View File

@@ -97,7 +97,7 @@ public class StreamToTableJoinIntegrationTests {
@StreamListener
@SendTo("output")
public KStream<String, Long> process(@Input("input") KStream<String, Long> userClicksStream,
@Input("inputX") KTable<String, String> userRegionsTable) {
@Input("input-x") KTable<String, String> userRegionsTable) {
return userClicksStream
.leftJoin(userRegionsTable, (clicks, region) -> new RegionWithClicks(region == null ? "UNKNOWN" : region, clicks),
@@ -111,7 +111,7 @@ public class StreamToTableJoinIntegrationTests {
interface KafkaStreamsProcessorX extends KafkaStreamsProcessor {
@Input("inputX")
@Input("input-x")
KTable<?, ?> inputX();
}
@@ -123,10 +123,10 @@ public class StreamToTableJoinIntegrationTests {
try (ConfigurableApplicationContext ignored = app.run("--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.input.destination=user-clicks",
"--spring.cloud.stream.bindings.inputX.destination=user-regions",
"--spring.cloud.stream.bindings.input-x.destination=user-regions",
"--spring.cloud.stream.bindings.output.destination=output-topic",
"--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.inputX.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.input-x.consumer.useNativeDecoding=true",
"--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde",
"--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde",

11
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java
View File

@@ -59,7 +59,9 @@ properties = {"spring.cloud.stream.kafka.bindings.output.producer.configuration.
"spring.cloud.stream.kafka.default.consumer.configuration.key.serializer=BarSerializer.class",
"spring.cloud.stream.kafka.default.consumer.configuration.value.serializer=BarSerializer.class",
"spring.cloud.stream.kafka.default.producer.configuration.foo=bar",
"spring.cloud.stream.kafka.bindings.output.producer.configuration.foo=bindingSpecificPropertyShouldWinOverDefault"})
"spring.cloud.stream.kafka.bindings.output.producer.configuration.foo=bindingSpecificPropertyShouldWinOverDefault",
"spring.cloud.stream.kafka.default.consumer.ackEachRecord=true",
"spring.cloud.stream.kafka.bindings.custom-in.consumer.ackEachRecord=false"})
public class KafkaBinderExtendedPropertiesTest {
private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";
@@ -122,6 +124,9 @@ public class KafkaBinderExtendedPropertiesTest {
//binding "input" gets BarSerializer and BarSerializer for ker.serializer/value.serializer through default properties.
assertThat(customKafkaConsumerProperties.getConfiguration().get("key.serializer")).isEqualTo("BarSerializer.class");
assertThat(customKafkaConsumerProperties.getConfiguration().get("value.serializer")).isEqualTo("BarSerializer.class");
assertThat(kafkaConsumerProperties.isAckEachRecord()).isEqualTo(true);
assertThat(customKafkaConsumerProperties.isAckEachRecord()).isEqualTo(false);
}
@EnableBinding(CustomBindingForExtendedPropertyTesting.class)
@@ -130,13 +135,13 @@ public class KafkaBinderExtendedPropertiesTest {
@StreamListener(Sink.INPUT)
@SendTo(Processor.OUTPUT)
public String process(String payload) throws InterruptedException {
public String process(String payload) {
return payload;
}
@StreamListener("custom-in")
@SendTo("custom-out")
public String processCustom(String payload) throws InterruptedException {
public String processCustom(String payload) {
return payload;
}