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

Remove usage of isOnlyLogRecordMetadata

Browse Source 
The `ConsumerProperties.isOnlyLogRecordMetadata` has been removed
in Spring for Apache Kafka in favor of `KafkaUtils.setConsumerRecordFormatter()`
This commit is contained in:
Artem Bilan
2022-03-15 13:48:07 -04:00
parent f3d3694836
commit 11c3acdf2c
1 changed files with 0 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -691,8 +691,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
private final boolean isSyncCommits;
private final boolean logOnlyMetadata;
private volatile boolean acknowledged;
private boolean autoAckEnabled = true;
@@ -716,7 +714,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
consumerProperties != null
? consumerProperties.getCommitLogLevel()
: LogIfLevelEnabled.Level.DEBUG);
this.logOnlyMetadata = consumerProperties != null && consumerProperties.isOnlyLogRecordMetadata();
}
@Override
@@ -761,7 +758,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
})
.collect(Collectors.toList());
if (rewound.size() > 0) {
KafkaUtils.setLogOnlyMetadata(this.logOnlyMetadata);
this.logger.warn(() -> "Rolled back " + KafkaUtils.format(record)
+ " later in-flight offsets "
+ rewound + " will also be re-fetched");
@@ -771,7 +767,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
}
private void commitIfPossible(ConsumerRecord<K, V> record) { // NOSONAR
KafkaUtils.setLogOnlyMetadata(this.logOnlyMetadata);
if (this.ackInfo.isRolledBack()) {
this.logger.warn(() -> "Cannot commit offset for "
+ KafkaUtils.format(record)