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

Update SIK to 3.2.0 snapshot

Browse Source 
Resolves #744

* Address deprecations in the pollable consumer
* Introduce a property for pollable time out in KafkaConsuerProperties
* Fix tests
* Add docs
* Addressin PR review comments
This commit is contained in:
Soby Chacko
2019-09-12 15:31:13 -04:00
committed by Artem Bilan
parent f2ab4a07c6
commit 6fdb0001d6
9 changed files with 74 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/src/main/asciidoc/overview.adoc
View File

@@ -265,6 +265,10 @@ The replication factor to use when provisioning topics. Overrides the binder-wid
Ignored if `replicas-assignments` is present.
+
Default: none (the binder-wide default of 1 is used).
pollTimeout::
Timeout used for polling in pollable consumers.
+
Default: 5 seconds.
==== Consuming Batches

7
pom.xml
View File

@@ -13,8 +13,9 @@
<properties>
<java.version>1.8</java.version>
<spring-kafka.version>2.3.0.BUILD-SNAPSHOT</spring-kafka.version>
<spring-integration-kafka.version>3.2.0.M4</spring-integration-kafka.version>
<spring-integration-kafka.version>3.2.0.BUILD-SNAPSHOT</spring-integration-kafka.version>
<kafka.version>2.3.0</kafka.version>
<spring-cloud-schema-registry.version>1.0.0.BUILD-SNAPSHOT</spring-cloud-schema-registry.version>
<spring-cloud-stream.version>3.0.0.BUILD-SNAPSHOT</spring-cloud-stream.version>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
<maven-checkstyle-plugin.failsOnViolation>true</maven-checkstyle-plugin.failsOnViolation>
@@ -113,8 +114,8 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-schema</artifactId>
<version>${spring-cloud-stream.version}</version>
<artifactId>spring-cloud-schema-registry-client</artifactId>
<version>${spring-cloud-schema-registry.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

14
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2019 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.
@@ -120,6 +120,11 @@ public class KafkaConsumerProperties {
private KafkaTopicProperties topic = new KafkaTopicProperties();
/**
* Timeout used for polling in pollable consumers.
*/
private long pollTimeout = org.springframework.kafka.listener.ConsumerProperties.DEFAULT_POLL_TIMEOUT;
public boolean isAckEachRecord() {
return this.ackEachRecord;
}
@@ -291,4 +296,11 @@ public class KafkaConsumerProperties {
this.topic = topic;
}
public long getPollTimeout() {
return this.pollTimeout;
}
public void setPollTimeout(long pollTimeout) {
this.pollTimeout = pollTimeout;
}
}

2
spring-cloud-stream-binder-kafka-streams/pom.xml
View File

@@ -88,7 +88,7 @@
<!-- Following dependencies are only provided for testing and won't be packaged with the binder apps-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-schema</artifactId>
<artifactId>spring-cloud-schema-registry-client</artifactId>
<scope>test</scope>
</dependency>
<dependency>

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

@@ -37,13 +37,13 @@ import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter;
import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.annotation.StreamMessageConverter;
import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor;
import org.springframework.cloud.stream.binder.kafka.streams.integration.utils.TestAvroSerializer;
import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -60,6 +60,7 @@ import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Soby Chacko
*/
@@ -174,9 +175,8 @@ public class PerRecordAvroContentTypeTests {
}
@Bean
@StreamMessageConverter
public MessageConverter sensorMessageConverter() throws IOException {
return new AvroSchemaMessageConverter();
return new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl());
}
}

5
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/utils/TestAvroSerializer.java
View File

@@ -21,7 +21,8 @@ import java.util.Map;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter;
import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
@@ -44,7 +45,7 @@ public class TestAvroSerializer<S> implements Serializer<S> {
@Override
public byte[] serialize(String topic, S data) {
AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter();
AvroSchemaMessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl());
Message<?> message = MessageBuilder.withPayload(data).build();
Map<String, Object> headers = new HashMap<>(message.getHeaders());
headers.put(MessageHeaders.CONTENT_TYPE, "application/avro");

5
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CompositeNonNativeSerdeTest.java
View File

@@ -27,8 +27,9 @@ import com.example.Sensor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Test;
import org.springframework.cloud.schema.registry.avro.AvroSchemaMessageConverter;
import org.springframework.cloud.schema.registry.avro.AvroSchemaServiceManagerImpl;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import static org.assertj.core.api.Assertions.assertThat;
@@ -51,7 +52,7 @@ public class CompositeNonNativeSerdeTest {
sensor.setTemperature(random.nextFloat() * 50);
List<MessageConverter> messageConverters = new ArrayList<>();
messageConverters.add(new AvroSchemaMessageConverter());
messageConverters.add(new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl()));
CompositeMessageConverterFactory compositeMessageConverterFactory = new CompositeMessageConverterFactory(
messageConverters, new ObjectMapper());
CompositeNonNativeSerde compositeNonNativeSerde = new CompositeNonNativeSerde(

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

@@ -97,6 +97,7 @@ import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
import org.springframework.kafka.listener.ConsumerProperties;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.kafka.support.KafkaHeaderMapper;
@@ -778,53 +779,34 @@ public class KafkaMessageChannelBinder extends
@Override
protected PolledConsumerResources createPolledConsumerResources(String name,
String group, ConsumerDestination destination,
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
boolean anonymous = !StringUtils.hasText(group);
Assert.isTrue(!anonymous || !consumerProperties.getExtension().isEnableDlq(),
final KafkaConsumerProperties extension = extendedConsumerProperties.getExtension();
Assert.isTrue(!anonymous || !extension.isEnableDlq(),
"DLQ support is not available for anonymous subscriptions");
String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString()
: group;
final ConsumerFactory<?, ?> consumerFactory = createKafkaConsumerFactory(
anonymous, consumerGroup, consumerProperties);
String[] topics = consumerProperties.isMultiplex()
anonymous, consumerGroup, extendedConsumerProperties);
String[] topics = extendedConsumerProperties.isMultiplex()
? StringUtils.commaDelimitedListToStringArray(destination.getName())
: new String[] { destination.getName() };
for (int i = 0; i < topics.length; i++) {
topics[i] = topics[i].trim();
}
KafkaMessageSource<?, ?> source = new KafkaMessageSource<>(consumerFactory,
topics);
source.setMessageConverter(getMessageConverter(consumerProperties));
source.setRawMessageHeader(consumerProperties.getExtension().isEnableDlq());
final ConsumerProperties consumerProperties = new ConsumerProperties(topics);
String clientId = name;
if (consumerProperties.getExtension().getConfiguration()
if (extension.getConfiguration()
.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)) {
clientId = consumerProperties.getExtension().getConfiguration()
clientId = extension.getConfiguration()
.get(ConsumerConfig.CLIENT_ID_CONFIG);
}
source.setClientId(clientId);
if (!consumerProperties.isMultiplex()) {
// I copied this from the regular consumer - it looks bogus to me - includes
// all partitions
// not just the ones this binding is listening to; doesn't seem right for a
// health check.
Collection<PartitionInfo> partitionInfos = getPartitionInfo(
destination.getName(), consumerProperties, consumerFactory, -1);
this.topicsInUse.put(destination.getName(),
new TopicInformation(consumerGroup, partitionInfos, false));
}
else {
for (int i = 0; i < topics.length; i++) {
Collection<PartitionInfo> partitionInfos = getPartitionInfo(topics[i],
consumerProperties, consumerFactory, -1);
this.topicsInUse.put(topics[i],
new TopicInformation(consumerGroup, partitionInfos, false));
}
}
consumerProperties.setClientId(clientId);
source.setRebalanceListener(new ConsumerRebalanceListener() {
consumerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
@@ -837,9 +819,36 @@ public class KafkaMessageChannelBinder extends
}
});
consumerProperties.setPollTimeout(extension.getPollTimeout());
KafkaMessageSource<?, ?> source = new KafkaMessageSource<>(consumerFactory,
consumerProperties);
source.setMessageConverter(getMessageConverter(extendedConsumerProperties));
source.setRawMessageHeader(extension.isEnableDlq());
if (!extendedConsumerProperties.isMultiplex()) {
// I copied this from the regular consumer - it looks bogus to me - includes
// all partitions
// not just the ones this binding is listening to; doesn't seem right for a
// health check.
Collection<PartitionInfo> partitionInfos = getPartitionInfo(
destination.getName(), extendedConsumerProperties, consumerFactory, -1);
this.topicsInUse.put(destination.getName(),
new TopicInformation(consumerGroup, partitionInfos, false));
}
else {
for (int i = 0; i < topics.length; i++) {
Collection<PartitionInfo> partitionInfos = getPartitionInfo(topics[i],
extendedConsumerProperties, consumerFactory, -1);
this.topicsInUse.put(topics[i],
new TopicInformation(consumerGroup, partitionInfos, false));
}
}
getMessageSourceCustomizer().configure(source, destination.getName(), group);
return new PolledConsumerResources(source, registerErrorInfrastructure(
destination, group, consumerProperties, true));
destination, group, extendedConsumerProperties, true));
}
@Override

2
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -2942,6 +2942,7 @@ public class KafkaBinderTests extends
this.messageConverter);
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProps = createConsumerProperties();
consumerProps.setMultiplex(true);
consumerProps.getExtension().setPollTimeout(1);
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer(
"pollable,anotherOne", "group-polledConsumer", inboundBindTarget,
consumerProps);
@@ -3028,6 +3029,7 @@ public class KafkaBinderTests extends
PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(
this.messageConverter);
ExtendedConsumerProperties<KafkaConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPollTimeout(1);
properties.setMaxAttempts(2);
properties.setBackOffInitialInterval(0);
properties.getExtension().setEnableDlq(true);