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

Fix S-K Deprecations

Browse Source
This commit is contained in:
Gary Russell
2020-04-02 13:52:33 -04:00
committed by Artem Bilan
parent 1e484d0fad
commit 213772dcaf
2 changed files with 8 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<K, V>
* @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<K, V> ackOnError(boolean ackOnError) {
this.target.getContainerProperties().setAckOnError(ackOnError);
return this;

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

@@ -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<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> 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<TopicPartition> 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())) {