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

Fix new and some old Sonar smells

Browse Source
This commit is contained in:
Artem Bilan
2020-07-14 10:48:23 -04:00
parent df1f260a59
commit a19e37237b
5 changed files with 189 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java
View File

@@ -49,7 +49,7 @@ public class KafkaTemplateSpec<K, V>
}
@Override
public KafkaTemplateSpec<K, V> id(String id) {
public KafkaTemplateSpec<K, V> id(String id) { // NOSONAR - visibility
return super.id(id);
}

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

@@ -32,7 +32,6 @@ import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.KafkaTemplate;
@@ -43,7 +42,6 @@ import org.springframework.kafka.listener.adapter.RecordMessagingMessageListener
import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.ConversionException;
import org.springframework.kafka.support.converter.KafkaMessageHeaders;
import org.springframework.kafka.support.converter.RecordMessageConverter;
@@ -110,7 +108,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
/**
* Set the message converter; must be a {@link RecordMessageConverter} or
* {@link BatchMessageConverter} depending on mode.
* {@link org.springframework.kafka.support.converter.BatchMessageConverter} depending on mode.
* @param messageConverter the converter.
*/
public void setMessageConverter(RecordMessageConverter messageConverter) {
@@ -216,7 +214,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
* attributes for use by the {@link ErrorMessageStrategy}.
* attributes for use by the {@link org.springframework.integration.support.ErrorMessageStrategy}.
* @param record the record.
* @param message the message.
*/

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

@@ -428,23 +428,11 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
if (this.paused && this.recordsIterator == null) {
this.logger.debug("Consumer is paused; no records will be returned");
}
ConsumerRecord<K, V> record;
if (this.recordsIterator != null) {
record = nextRecord();
}
else {
synchronized (this.consumerMonitor) {
ConsumerRecords<K, V> records = this.consumer
.poll(this.assignedPartitions.isEmpty() ? this.assignTimeout : this.pollTimeout);
if (records == null || records.count() == 0) {
return null;
}
this.remainingCount.set(records.count());
this.recordsIterator = records.iterator();
record = nextRecord();
}
}
return recordToMessage(record);
ConsumerRecord<K, V> record = pollRecord();
return record != null
? recordToMessage(record)
: null;
}
protected void createConsumer() {
@@ -461,48 +449,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
this.consumer.subscribe(topicPattern, rebalanceCallback);
}
else if (partitions != null) {
List<TopicPartition> topicPartitionsToAssign =
Arrays.stream(partitions)
.map(TopicPartitionOffset::getTopicPartition)
.collect(Collectors.toList());
this.consumer.assign(topicPartitionsToAssign);
this.assignedPartitions.addAll(topicPartitionsToAssign);
for (TopicPartitionOffset partition : partitions) {
if (TopicPartitionOffset.SeekPosition.BEGINNING.equals(partition.getPosition())) {
this.consumer.seekToBeginning(Collections.singleton(partition.getTopicPartition()));
}
else if (TopicPartitionOffset.SeekPosition.END.equals(partition.getPosition())) {
this.consumer.seekToEnd(Collections.singleton(partition.getTopicPartition()));
}
else {
TopicPartition topicPartition = partition.getTopicPartition();
Long offset = partition.getOffset();
if (offset != null) {
long newOffset = offset;
if (offset < 0) {
if (!partition.isRelativeToCurrent()) {
this.consumer.seekToEnd(Collections.singleton(topicPartition));
continue;
}
newOffset = Math.max(0, this.consumer.position(topicPartition) + offset);
}
else if (partition.isRelativeToCurrent()) {
newOffset = this.consumer.position(topicPartition) + offset;
}
try {
this.consumer.seek(topicPartition, newOffset);
}
catch (Exception e) {
this.logger.error("Failed to set initial offset for " + topicPartition
+ " at " + newOffset + ". Position is " + this.consumer
.position(topicPartition), e);
}
}
}
}
assignAndSeekPartitionts(partitions);
}
else {
this.consumer.subscribe(Arrays.asList(this.consumerProperties.getTopics()), // NOSONAR
@@ -511,6 +458,70 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
}
}
private void assignAndSeekPartitionts(TopicPartitionOffset[] partitions) {
List<TopicPartition> topicPartitionsToAssign =
Arrays.stream(partitions)
.map(TopicPartitionOffset::getTopicPartition)
.collect(Collectors.toList());
this.consumer.assign(topicPartitionsToAssign);
this.assignedPartitions.addAll(topicPartitionsToAssign);
for (TopicPartitionOffset partition : partitions) {
if (TopicPartitionOffset.SeekPosition.BEGINNING.equals(partition.getPosition())) {
this.consumer.seekToBeginning(Collections.singleton(partition.getTopicPartition()));
}
else if (TopicPartitionOffset.SeekPosition.END.equals(partition.getPosition())) {
this.consumer.seekToEnd(Collections.singleton(partition.getTopicPartition()));
}
else {
TopicPartition topicPartition = partition.getTopicPartition();
Long offset = partition.getOffset();
if (offset != null) {
long newOffset = offset;
if (offset < 0) {
if (!partition.isRelativeToCurrent()) {
this.consumer.seekToEnd(Collections.singleton(topicPartition));
continue;
}
newOffset = Math.max(0, this.consumer.position(topicPartition) + offset);
}
else if (partition.isRelativeToCurrent()) {
newOffset = this.consumer.position(topicPartition) + offset;
}
try {
this.consumer.seek(topicPartition, newOffset);
}
catch (Exception e) {
this.logger.error("Failed to set initial offset for " + topicPartition
+ " at " + newOffset + ". Position is " + this.consumer
.position(topicPartition), e);
}
}
}
}
}
@Nullable
private ConsumerRecord<K, V> pollRecord() {
if (this.recordsIterator != null) {
return nextRecord();
}
else {
synchronized (this.consumerMonitor) {
ConsumerRecords<K, V> records = this.consumer
.poll(this.assignedPartitions.isEmpty() ? this.assignTimeout : this.pollTimeout);
if (records == null || records.count() == 0) {
return null;
}
this.remainingCount.set(records.count());
this.recordsIterator = records.iterator();
return nextRecord();
}
}
}
private ConsumerRecord<K, V> nextRecord() {
ConsumerRecord<K, V> record;
record = this.recordsIterator.next();

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -566,8 +566,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
ListenableFuture<SendResult<K, V>> future, MessageChannel metadataChannel)
throws InterruptedException, ExecutionException {
final MessageChannel sendFailureChannel = getSendFailureChannel();
if (sendFailureChannel != null || metadataChannel != null) {
final MessageChannel failureChannel = getSendFailureChannel();
if (failureChannel != null || metadataChannel != null) {
future.addCallback(new ListenableFutureCallback<SendResult<K, V>>() { // NOSONAR
@Override
@@ -581,8 +581,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
@Override
public void onFailure(Throwable ex) {
if (sendFailureChannel != null) {
KafkaProducerMessageHandler.this.messagingTemplate.send(sendFailureChannel,
if (failureChannel != null) {
KafkaProducerMessageHandler.this.messagingTemplate.send(failureChannel,
KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage(
new KafkaSendFailureException(message, producerRecord, ex), null));
}