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

Use LogAccessor from SF

Browse Source 
* Change main classes to use a `LogAccessor` API to simplify code flow
* Fix tests according `LogAccessor` property
* Fix some Sonar smells
This commit is contained in:
Artem Bilan
2020-10-06 13:56:50 -04:00
parent a66e82b0aa
commit c7ff99a4e8
95 changed files with 1165 additions and 1432 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java
View File

@@ -20,7 +20,6 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.core.log.LogAccessor;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.support.KafkaHeaders;
@@ -38,8 +37,6 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractKafkaChannel extends AbstractMessageChannel {
protected final LogAccessor logger = new LogAccessor(super.logger); // NOSONAR final
private final KafkaOperations<?, ?> template;
protected final String topic; // NOSONAR final
@@ -74,8 +71,8 @@ public abstract class AbstractKafkaChannel extends AbstractMessageChannel {
protected boolean doSend(Message<?> message, long timeout) {
try {
this.template.send(MessageBuilder.fromMessage(message)
.setHeader(KafkaHeaders.TOPIC, this.topic)
.build())
.setHeader(KafkaHeaders.TOPIC, this.topic)
.build())
.get(timeout < 0 ? Long.MAX_VALUE : timeout, TimeUnit.MILLISECONDS);
}
catch (@SuppressWarnings("unused") InterruptedException e) {

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

@@ -348,11 +348,9 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
throw new IllegalArgumentException("Custom consumer factory is not configured with '"
+ ConsumerConfig.MAX_POLL_RECORDS_CONFIG + " = 1'");
}
if (this.logger.isWarnEnabled()) {
this.logger.warn("'" + ConsumerConfig.MAX_POLL_RECORDS_CONFIG
+ "' has been forced from " + (maxPoll == null ? "unspecified" : maxPoll)
+ " to 1, to avoid having to seek after each record");
}
this.logger.warn(() -> ConsumerConfig.MAX_POLL_RECORDS_CONFIG
+ "' has been forced from " + (maxPoll == null ? "unspecified" : maxPoll)
+ " to 1, to avoid having to seek after each record");
Map<String, Object> configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties());
configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
DefaultKafkaConsumerFactory<K, V> fixedConsumerFactory = new DefaultKafkaConsumerFactory<>(configs);
@@ -450,7 +448,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
this.consumer.subscribe(topicPattern, rebalanceCallback);
}
else if (partitions != null) {
assignAndSeekPartitionts(partitions);
assignAndSeekPartitions(partitions);
}
else {
this.consumer.subscribe(Arrays.asList(this.consumerProperties.getTopics()), // NOSONAR
@@ -459,7 +457,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
}
}
private void assignAndSeekPartitionts(TopicPartitionOffset[] partitions) {
private void assignAndSeekPartitions(TopicPartitionOffset[] partitions) {
List<TopicPartition> topicPartitionsToAssign =
Arrays.stream(partitions)
.map(TopicPartitionOffset::getTopicPartition)
@@ -478,7 +476,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
TopicPartition topicPartition = partition.getTopicPartition();
Long offset = partition.getOffset();
if (offset != null) {
long newOffset = offset;
long newOffset;
if (offset < 0) {
if (!partition.isRelativeToCurrent()) {
@@ -490,14 +488,17 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
else if (partition.isRelativeToCurrent()) {
newOffset = this.consumer.position(topicPartition) + offset;
}
else {
newOffset = offset;
}
try {
this.consumer.seek(topicPartition, newOffset);
}
catch (Exception e) {
this.logger.error("Failed to set initial offset for " + topicPartition
catch (Exception ex) {
this.logger.error(ex, () -> "Failed to set initial offset for " + topicPartition
+ " at " + newOffset + ". Position is " + this.consumer
.position(topicPartition), e);
.position(topicPartition));
}
}
}
@@ -759,8 +760,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
return i.getRecord().offset();
})
.collect(Collectors.toList());
if (rewound.size() > 0 && this.logger.isWarnEnabled()) {
this.logger.warn("Rolled back " + ListenerUtils.recordToString(record, this.logOnlyMetadata)
if (rewound.size() > 0) {
this.logger.warn(() -> "Rolled back " + ListenerUtils.recordToString(record, this.logOnlyMetadata)
+ " later in-flight offsets "
+ rewound + " will also be re-fetched");
}
@@ -770,11 +771,9 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
private void commitIfPossible(ConsumerRecord<K, V> record) { // NOSONAR
if (this.ackInfo.isRolledBack()) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("Cannot commit offset for "
+ ListenerUtils.recordToString(record, this.logOnlyMetadata)
+ "; an earlier offset was rolled back");
}
this.logger.warn(() -> "Cannot commit offset for "
+ ListenerUtils.recordToString(record, this.logOnlyMetadata)
+ "; an earlier offset was rolled back");
}
else {
Set<KafkaAckInfo<K, V>> candidates = this.ackInfo.getOffsets().get(this.ackInfo.getTopicPartition());

9
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java
View File

@@ -69,6 +69,7 @@ import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.core.log.LogMessage;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.StaticMessageHeaderAccessor;
import org.springframework.integration.acks.AcknowledgmentCallback;
@@ -581,15 +582,15 @@ class MessageSourceTests {
inOrder.verify(consumer).poll(any(Duration.class));
inOrder.verify(consumer).seek(topicPartition, 0L); // rollback
inOrder.verify(log1).isWarnEnabled();
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
ArgumentCaptor<LogMessage> captor = ArgumentCaptor.forClass(LogMessage.class);
inOrder.verify(log1).warn(captor.capture());
assertThat(captor.getValue())
assertThat(captor.getValue().toString())
.contains("Rolled back")
.contains("later in-flight offsets [1] will also be re-fetched");
inOrder.verify(log2).isWarnEnabled();
captor = ArgumentCaptor.forClass(String.class);
captor = ArgumentCaptor.forClass(LogMessage.class);
inOrder.verify(log2).warn(captor.capture());
assertThat(captor.getValue())
assertThat(captor.getValue().toString())
.contains("Cannot commit offset for ConsumerRecord")
.contains("; an earlier offset was rolled back");
inOrder.verify(consumer).poll(any(Duration.class));