Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2021 the original author or authors.
* Copyright 2021-2023 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.
Expand Down Expand Up @@ -37,6 +37,8 @@ public interface BatchInterceptor<K, V> extends ThreadStateProcessor {
/**
* Perform some action on the records or return a different one. If null is returned
* the records will be skipped. Invoked before the listener.
* IMPORTANT: If transactions are being used, and this method throws an exception, it
* cannot be used with the container's {@code interceptBeforeTx} property set to true.
* @param records the records.
* @param consumer the consumer.
* @return the records or null.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -714,24 +714,24 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
private final Duration syncCommitTimeout;

private final RecordInterceptor<K, V> recordInterceptor =
!isInterceptBeforeTx() && this.transactionManager != null
!isInterceptBeforeTx() || this.transactionManager == null
? getRecordInterceptor()
: null;

private final RecordInterceptor<K, V> earlyRecordInterceptor =
isInterceptBeforeTx() || this.transactionManager == null
isInterceptBeforeTx() && this.transactionManager != null
? getRecordInterceptor()
: null;

private final RecordInterceptor<K, V> commonRecordInterceptor = getRecordInterceptor();

private final BatchInterceptor<K, V> batchInterceptor =
!isInterceptBeforeTx() && this.transactionManager != null
!isInterceptBeforeTx() || this.transactionManager == null
? getBatchInterceptor()
: null;

private final BatchInterceptor<K, V> earlyBatchInterceptor =
isInterceptBeforeTx() || this.transactionManager == null
isInterceptBeforeTx() && this.transactionManager != null
? getBatchInterceptor()
: null;

Expand Down Expand Up @@ -2915,7 +2915,6 @@ private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {
if (cRecord == null) {
this.logger.debug(() -> "RecordInterceptor returned null, skipping: "
+ KafkaUtils.format(recordArg));
ackCurrent(recordArg);
}
else {
try {
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 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.
Expand Down Expand Up @@ -38,8 +38,11 @@ public interface RecordInterceptor<K, V> extends ThreadStateProcessor {
/**
* Perform some action on the record or return a different one. If null is returned
* the record will be skipped. Invoked before the listener. IMPORTANT; if this method
* returns a different record, the topic, partition and offset must not be changed
* to avoid undesirable side-effects.
* returns a different record, the topic, partition and offset must not be changed to
* avoid undesirable side-effects.
* <p>
* IMPORTANT: If transactions are being used, and this method throws an exception, it
* cannot be used with the container's {@code interceptBeforeTx} property set to true.
* @param record the record.
* @param consumer the consumer.
* @return the record or null.
Expand Down