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

Resolve deprecations and other compiler warnings

Browse Source
This commit is contained in:
Gary Russell
2019-07-17 16:30:08 -04:00
committed by Artem Bilan
parent 600e88ec56
commit 41de7495e6
7 changed files with 84 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.kafka.dsl;
import java.util.Arrays;
import java.util.regex.Pattern;
import org.apache.kafka.common.TopicPartition;
@@ -29,7 +30,7 @@ import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.GenericMessageListenerContainer;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.TopicPartitionOffset;
/**
* Factory class for Apache Kafka components.
@@ -214,6 +215,26 @@ public final class Kafka {
containerProperties), listenerMode);
}
/**
* Create an initial
* {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
* @param consumerFactory the {@link ConsumerFactory}.
* @param topicPartitions the {@link TopicPartition} vararg.
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type.
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
* @deprecated in favor of {@link #messageDrivenChannelAdapter(ConsumerFactory, TopicPartitionOffset...)}.
*/
@Deprecated
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
org.springframework.kafka.support.TopicPartitionInitialOffset... topicPartitions) {
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
topicPartitions);
}
/**
* Create an initial
* {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
@@ -226,12 +247,39 @@ public final class Kafka {
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
TopicPartitionInitialOffset... topicPartitions) {
TopicPartitionOffset... topicPartitions) {
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
topicPartitions);
}
/**
* Create an initial
* {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
* @param consumerFactory the {@link ConsumerFactory}.
* @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
* @param topicPartitions the {@link TopicPartition} vararg.
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type.
* @return the
* KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
* @deprecated in favor of
* {@link #messageDrivenChannelAdapter(ConsumerFactory, org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.ListenerMode, TopicPartitionOffset...)}
*/
@Deprecated
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode,
org.springframework.kafka.support.TopicPartitionInitialOffset... topicPartitions) {
return messageDrivenChannelAdapter(
new KafkaMessageListenerContainerSpec<>(consumerFactory,
Arrays.stream(topicPartitions)
.map(org.springframework.kafka.support.TopicPartitionInitialOffset::toTPO)
.toArray(TopicPartitionOffset[]::new)), listenerMode);
}
/**
* Create an initial
* {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
@@ -246,7 +294,7 @@ public final class Kafka {
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode,
TopicPartitionInitialOffset... topicPartitions) {
TopicPartitionOffset... topicPartitions) {
return messageDrivenChannelAdapter(
new KafkaMessageListenerContainerSpec<>(consumerFactory,

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java
View File

@@ -123,8 +123,7 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
* @param <R> the reply value type.
*/
public static class KafkaInboundGatewayListenerContainerSpec<K, V, R> extends
KafkaInboundGatewaySpec<K, V, R, KafkaInboundGatewayListenerContainerSpec<K, V, R>>
implements ComponentsRegistration {
KafkaInboundGatewaySpec<K, V, R, KafkaInboundGatewayListenerContainerSpec<K, V, R>> {
private final KafkaMessageListenerContainerSpec<K, V> containerSpec;

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java
View File

@@ -149,7 +149,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
* @return the spec
* @since 3.2.0
*/
public S payloadType(Class payloadType) {
public S payloadType(Class<?> payloadType) {
this.target.setPayloadType(payloadType);
return _this();
}
@@ -199,8 +199,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
* @param <V> the value type.
*/
public static class KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> extends
KafkaMessageDrivenChannelAdapterSpec<K, V, KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>>
implements ComponentsRegistration {
KafkaMessageDrivenChannelAdapterSpec<K, V, KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>> {
private final KafkaMessageListenerContainerSpec<K, V> spec;

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java
View File

@@ -30,7 +30,7 @@ import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.GenericErrorHandler;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.TopicPartitionOffset;
/**
* A helper class in the Builder pattern style to delegate options to the
@@ -54,7 +54,7 @@ public class KafkaMessageListenerContainerSpec<K, V>
}
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory,
TopicPartitionInitialOffset... topicPartitions) {
TopicPartitionOffset... topicPartitions) {
this(consumerFactory, new ContainerProperties(topicPartitions));
}

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
View File

@@ -170,13 +170,13 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
@Override
protected void onInit() {
super.onInit();
MessageListener<K, V> listener = this.listener;
MessageListener<K, V> kafkaListener = this.listener;
if (this.retryTemplate != null) {
listener = new RetryingMessageListenerAdapter<>(listener, this.retryTemplate,
kafkaListener = new RetryingMessageListenerAdapter<>(kafkaListener, this.retryTemplate,
this.recoveryCallback);
this.retryTemplate.registerListener(this.listener);
}
this.messageListenerContainer.getContainerProperties().setMessageListener(listener);
this.messageListenerContainer.getContainerProperties().setMessageListener(kafkaListener);
}
@Override

37
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java
View File

@@ -96,22 +96,22 @@ public class InboundGatewayTests {
@Test
public void testInbound() throws Exception {
EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
EmbeddedKafkaBroker embedded = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps("replyHandler1", "false", embeddedKafka);
KafkaTestUtils.consumerProps("replyHandler1", "false", embedded);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf2.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic2);
embedded.consumeFromAnEmbeddedTopic(consumer, topic2);
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "false", embedded);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic1);
containerProps.setIdleEventInterval(100L);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embedded);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic1);
@@ -127,8 +127,9 @@ public class InboundGatewayTests {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, consumer, type);
Consumer<?, ?> con, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, con, type);
return MessageBuilder.fromMessage(message)
.setHeader("testHeader", "testValue")
.setHeader(KafkaHeaders.REPLY_TOPIC, topic2)
@@ -170,7 +171,7 @@ public class InboundGatewayTests {
}
@Test
public void testInboundErrorRecover() throws Exception {
public void testInboundErrorRecover() {
EmbeddedKafkaBroker broker = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", broker);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
@@ -205,8 +206,8 @@ public class InboundGatewayTests {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, consumer, type);
Consumer<?, ?> con, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, con, type);
return MessageBuilder.fromMessage(message)
.setHeader("testHeader", "testValue")
.setHeader(KafkaHeaders.REPLY_TOPIC, topic4)
@@ -250,21 +251,21 @@ public class InboundGatewayTests {
}
@Test
public void testInboundRetryErrorRecover() throws Exception {
EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embeddedKafka);
public void testInboundRetryErrorRecover() {
EmbeddedKafkaBroker embedded = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embedded);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf2.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic6);
embedded.consumeFromAnEmbeddedTopic(consumer, topic6);
Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "false", embedded);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic5);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embedded);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic5);
@@ -285,8 +286,8 @@ public class InboundGatewayTests {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, consumer, type);
Consumer<?, ?> con, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, con, type);
return MessageBuilder.fromMessage(message)
.setHeader("testHeader", "testValue")
.setHeader(KafkaHeaders.REPLY_TOPIC, topic6)

12
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -69,7 +69,7 @@ import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.TopicPartitionOffset;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
import org.springframework.kafka.support.converter.ConversionException;
@@ -119,7 +119,7 @@ public class MessageDrivenAdapterTests {
private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
@Test
public void testInboundRecord() throws Exception {
public void testInboundRecord() {
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -204,7 +204,7 @@ public class MessageDrivenAdapterTests {
}
@Test
public void testInboundRecordRetryRecover() throws Exception {
public void testInboundRecordRetryRecover() {
Map<String, Object> props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -257,7 +257,7 @@ public class MessageDrivenAdapterTests {
}
@Test
public void testInboundRecordNoRetryRecover() throws Exception {
public void testInboundRecordNoRetryRecover() {
Map<String, Object> props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -515,8 +515,8 @@ public class MessageDrivenAdapterTests {
resumeLatch.countDown();
return null;
}).given(consumer).resume(records.keySet());
TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] {
new TopicPartitionInitialOffset("foo", 0) };
TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
new TopicPartitionOffset("foo", 0) };
ContainerProperties containerProps = new ContainerProperties(topicPartition);
containerProps.setAckMode(ContainerProperties.AckMode.RECORD);
containerProps.setClientId("clientId");