Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

KMessageSource getter for consumer properties

Browse Source 
- allow customization after creation - e.g. SCSt `MessageSourceCustomizer`
- also fix deprecation in `ReplyingKafkaTemplate`
This commit is contained in:
Gary Russell
2019-09-25 15:18:28 -04:00
committed by Artem Bilan
parent adf91edf05
commit 8cc2b2cc00
4 changed files with 36 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.kafka.dsl;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.function.Consumer;
@@ -119,14 +120,32 @@ public class KafkaOutboundGatewaySpec<K, V, R, S extends KafkaOutboundGatewaySpe
}
@SuppressWarnings("unchecked")
ReplyingKafkaTemplateSpec<K, V, R> taskScheduler(TaskScheduler scheduler) {
public ReplyingKafkaTemplateSpec<K, V, R> taskScheduler(TaskScheduler scheduler) {
((ReplyingKafkaTemplate<K, V, R>) this.target).setTaskScheduler(scheduler);
return this;
}
/**
* Default reply timeout.
* @param replyTimeout the timeout.
* @return the spec.
* @deprecated in favor of {@link #defaultReplyTimeout(Duration)}.
*/
@Deprecated
@SuppressWarnings("unchecked")
ReplyingKafkaTemplateSpec<K, V, R> replyTimeout(long replyTimeout) {
((ReplyingKafkaTemplate<K, V, R>) this.target).setReplyTimeout(replyTimeout);
public ReplyingKafkaTemplateSpec<K, V, R> replyTimeout(long replyTimeout) {
((ReplyingKafkaTemplate<K, V, R>) this.target).setDefaultReplyTimeout(Duration.ofMillis(replyTimeout));
return this;
}
/**
* Default reply timeout.
* @param replyTimeout the timeout.
* @return the spec.
*/
@SuppressWarnings("unchecked")
public ReplyingKafkaTemplateSpec<K, V, R> defaultReplyTimeout(Duration replyTimeout) {
((ReplyingKafkaTemplate<K, V, R>) this.target).setDefaultReplyTimeout(replyTimeout);
return this;
}

10
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
View File

@@ -270,6 +270,16 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
}
}
/**
* Get a reference to the configured consumer properties; allows further
* customization of the properties before the source is started.
* @return the properties.
* @since 3.2
*/
public ConsumerProperties getConsumerProperties() {
return this.consumerProperties;
}
protected String getGroupId() {
return this.consumerProperties.getGroupId();
}

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.kafka.dsl;
import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
@@ -354,7 +355,7 @@ public class KafkaDslTests {
return IntegrationFlows.from(Gate.class)
.handle(Kafka.outboundGateway(producerFactory(), replyContainer())
.sync(true)
.configureKafkaTemplate(t -> t.replyTimeout(30_000)))
.configureKafkaTemplate(t -> t.defaultReplyTimeout(Duration.ofSeconds(30))))
.get();
}

3
spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt
View File

@@ -73,6 +73,7 @@ import org.springframework.messaging.support.GenericMessage
import org.springframework.retry.support.RetryTemplate
import org.springframework.test.annotation.DirtiesContext
import org.springframework.test.context.junit4.SpringRunner
import java.time.Duration
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import java.util.stream.Stream
@@ -328,7 +329,7 @@ class KafkaDslKotlinTests {
fun replyingKafkaTemplate() =
ReplyingKafkaTemplate(producerFactory(), replyContainer())
.also {
it.setReplyTimeout(30000)
it.setDefaultReplyTimeout(Duration.ofSeconds(30))
}
@Bean