From 213772dcaffb406ab0c4648dddfb5ab45f584fbf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 2 Apr 2020 13:52:33 -0400 Subject: [PATCH] Fix S-K Deprecations --- .../kafka/dsl/KafkaMessageListenerContainerSpec.java | 4 +++- .../integration/kafka/inbound/KafkaMessageSource.java | 10 +++++----- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java index 7b4e869678..de596ece25 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -235,8 +235,10 @@ public class KafkaMessageListenerContainerSpec * @param ackOnError whether the container should acknowledge messages that throw * exceptions. * @return the spec. + * @deprecated in favor of {@code GenericErrorHandler.isAckAfterHandle()}. * @see ContainerProperties#setAckOnError(boolean) */ + @Deprecated public KafkaMessageListenerContainerSpec ackOnError(boolean ackOnError) { this.target.getContainerProperties().setAckOnError(ackOnError); return this; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index aa12332d63..6e2510a417 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -220,7 +220,7 @@ public class KafkaMessageSource extends AbstractMessageSource impl Assert.notNull(ackCallbackFactory, "'ackCallbackFactory' must not be null"); Assert.isTrue( !ObjectUtils.isEmpty(consumerProperties.getTopics()) - || !ObjectUtils.isEmpty(consumerProperties.getTopicPartitionsToAssign()) + || !ObjectUtils.isEmpty(consumerProperties.getTopicPartitions()) || consumerProperties.getTopicPattern() != null, "topics, topicPattern, or topicPartitions must be provided" ); @@ -518,15 +518,15 @@ public class KafkaMessageSource extends AbstractMessageSource impl if (this.consumerProperties.getTopicPattern() != null) { this.consumer.subscribe(this.consumerProperties.getTopicPattern(), rebalanceCallback); } - else if (this.consumerProperties.getTopicPartitionsToAssign() != null) { + else if (this.consumerProperties.getTopicPartitions() != null) { List topicPartitionsToAssign = Arrays - .stream(this.consumerProperties.getTopicPartitionsToAssign()) + .stream(this.consumerProperties.getTopicPartitions()) .map(TopicPartitionOffset::getTopicPartition) .collect(Collectors.toList()); this.consumer.assign(topicPartitionsToAssign); this.assignedPartitions = new ArrayList<>(topicPartitionsToAssign); - TopicPartitionOffset[] partitions = this.consumerProperties.getTopicPartitionsToAssign(); + TopicPartitionOffset[] partitions = this.consumerProperties.getTopicPartitions(); for (TopicPartitionOffset partition : partitions) { if (TopicPartitionOffset.SeekPosition.BEGINNING.equals(partition.getPosition())) {