GH-2588: ARBP support BatchListenerFailedException

Zhiyang.Wang1 · Zhiyang.Wang1 · commit ec7cde6eebed · 2023-11-09T16:13:25.000+08:00
Resolves #2588 * add default function `processBatch` in AfterRollbackProcessor * support BatchListenerFailedException in DefaultAfterRollbackProcessor * Add unit tests
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AfterRollbackProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AfterRollbackProcessor.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-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.
@@ -20,6 +20,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 
 import org.springframework.kafka.listener.ContainerProperties.EOSMode;
 
@@ -63,6 +64,12 @@ public interface AfterRollbackProcessor<K, V> {
 	void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer,
 			MessageListenerContainer container, Exception exception, boolean recoverable, EOSMode eosMode);
 
+	default void processBatch(ConsumerRecords<K, V> records, List<ConsumerRecord<K, V>> recordList,
+			Consumer<K, V> consumer, MessageListenerContainer container, Exception exception,
+			boolean recoverable, ContainerProperties.EOSMode eosMode) {
+		process(recordList, consumer, container, exception, recoverable, eosMode);
+	}
+
 	/**
 	 * Optional method to clear thread state; will be called just before a consumer
 	 * thread terminates.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchListenerFailedException.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchListenerFailedException.java
@@ -98,9 +98,9 @@ public int getIndex() {
 
 	@Override
 	public String getMessage() {
-		return super.getMessage() + (this.record != null
+		return super.getMessage() + " " + (this.record != null
 				? (this.record.topic() + "-" + this.record.partition() + "@" + this.record.offset())
-				: (" @-" + this.index));
+				: ("@-" + this.index));
 	}
 
 }
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java
@@ -16,14 +16,17 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BiConsumer;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 
@@ -51,7 +54,7 @@
  * @since 1.3.5
  *
  */
-public class DefaultAfterRollbackProcessor<K, V> extends FailedRecordProcessor
+public class DefaultAfterRollbackProcessor<K, V> extends FailedBatchProcessor
 		implements AfterRollbackProcessor<K, V> {
 
 	private final Map<Thread, BackOffExecution> backOffs = new ConcurrentHashMap<>();
@@ -138,7 +141,7 @@ public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>,
 			BackOff backOff, @Nullable BackOffHandler backOffHandler, @Nullable KafkaOperations<?, ?> kafkaOperations,
 			boolean commitRecovered) {
 
-		super(recoverer, backOff, backOffHandler);
+		super(recoverer, backOff, backOffHandler, null);
 		this.kafkaTemplate = kafkaOperations;
 		super.setCommitRecovered(commitRecovered);
 		checkConfig();
@@ -176,6 +179,46 @@ && isCommitRecovered() && this.kafkaTemplate.isTransactional()) {
 
 	}
 
+	@Override
+	public void processBatch(ConsumerRecords<K, V> records, List<ConsumerRecord<K, V>> recordList, Consumer<K, V> consumer,
+			@Nullable MessageListenerContainer container, Exception exception, boolean recoverable, EOSMode eosMode) {
+
+		int index = handlerBatchListenerFailedException(exception, records);
+		if (index > -1) {
+			List<ConsumerRecord<?, ?>> toCommit = new ArrayList<>();
+			List<ConsumerRecord<?, ?>> remaining = new ArrayList<>();
+			for (ConsumerRecord<?, ?> datum : records) {
+				if (index-- > 0) {
+					toCommit.add(datum);
+				}
+				else {
+					remaining.add(datum);
+				}
+			}
+			Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+			toCommit.forEach(rec -> offsets.compute(new TopicPartition(rec.topic(), rec.partition()),
+					(key, val) -> ListenerUtils.createOffsetAndMetadata(container, rec.offset() + 1)));
+			if (offsets.size() > 0 && this.kafkaTemplate != null && this.kafkaTemplate.isTransactional()) {
+				this.kafkaTemplate.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
+			}
+
+			// skip first record
+			if (SeekUtils.doSeeks(remaining, consumer, exception, recoverable,
+					getFailureTracker()::recovered, container, this.logger)
+						&& isCommitRecovered() && this.kafkaTemplate.isTransactional()) {
+				ConsumerRecord<?, ?> skipped = remaining.get(0);
+				this.kafkaTemplate.sendOffsetsToTransaction(
+						Collections.singletonMap(new TopicPartition(skipped.topic(), skipped.partition()),
+								createOffsetAndMetadata(container, skipped.offset() + 1)
+						), consumer.groupMetadata());
+			}
+		}
+		else {
+			process(recordList, consumer, container, exception, false, eosMode);
+		}
+
+	}
+
 	@Override
 	public boolean isProcessInTransaction() {
 		return isCommitRecovered();
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedBatchProcessor.java
@@ -120,7 +120,7 @@ public void setReclassifyOnExceptionChange(boolean reclassifyOnExceptionChange)
 	@Override
 	protected void notRetryable(Stream<Class<? extends Exception>> notRetryable) {
 		if (this.fallbackBatchHandler instanceof ExceptionClassifier handler) {
-			notRetryable.forEach(ex -> handler.addNotRetryableExceptions(ex));
+			notRetryable.forEach(handler::addNotRetryableExceptions);
 		}
 	}
 
@@ -159,11 +159,19 @@ protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> data, C
 
 	protected <K, V> ConsumerRecords<K, V> handle(Exception thrownException, ConsumerRecords<?, ?> data,
 			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
+		int index = handlerBatchListenerFailedException(thrownException, data);
+		if (index > -1) {
+			return seekOrRecover(thrownException, data, consumer, container, index);
+		}
+		fallback(thrownException, data, consumer, container, invokeListener);
+		return ConsumerRecords.empty();
+	}
+
+	protected int handlerBatchListenerFailedException(Exception thrownException, ConsumerRecords<?, ?> data) {
 
 		BatchListenerFailedException batchListenerFailedException = getBatchListenerFailedException(thrownException);
 		if (batchListenerFailedException == null) {
 			this.logger.debug(thrownException, "Expected a BatchListenerFailedException; re-delivering full batch");
-			fallback(thrownException, data, consumer, container, invokeListener);
 		}
 		else {
 			getRetryListeners().forEach(listener -> listener.failedDelivery(data, thrownException, 1));
@@ -181,13 +189,12 @@ protected <K, V> ConsumerRecords<K, V> handle(Exception thrownException, Consume
 
 					}
 				});
-				fallback(thrownException, data, consumer, container, invokeListener);
 			}
 			else {
-				return seekOrRecover(thrownException, data, consumer, container, index);
+				return index;
 			}
 		}
-		return ConsumerRecords.empty();
+		return -1;
 	}
 
 	private void fallback(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
@@ -201,11 +208,9 @@ private int findIndex(ConsumerRecords<?, ?> data, ConsumerRecord<?, ?> record) {
 			return -1;
 		}
 		int i = 0;
-		Iterator<?> iterator = data.iterator();
-		while (iterator.hasNext()) {
-			ConsumerRecord<?, ?> candidate = (ConsumerRecord<?, ?>) iterator.next();
-			if (candidate.topic().equals(record.topic()) && candidate.partition() == record.partition()
-					&& candidate.offset() == record.offset()) {
+		for (ConsumerRecord<?, ?> datum : data) {
+			if (datum.topic().equals(record.topic()) && datum.partition() == record.partition()
+					&& datum.offset() == record.offset()) {
 				break;
 			}
 			i++;
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
@@ -916,7 +917,6 @@ else if (listener instanceof MessageListener) {
 			if (this.containerProperties.isLogContainerConfig()) {
 				this.logger.info(toString());
 			}
-			Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
 			ApplicationContext applicationContext = getApplicationContext();
 			this.checkNullKeyForExceptions = this.containerProperties.isCheckDeserExWhenKeyNull()
 					|| ErrorHandlingUtils.checkDeserializer(KafkaMessageListenerContainer.this.consumerFactory,
@@ -2180,7 +2180,11 @@ private void batchRollback(final ConsumerRecords<K, V> records,
 
 					@Override
 					protected void doInTransactionWithoutResult(TransactionStatus status) {
-						batchAfterRollback(records, recordList, e, afterRollbackProcessorToUse);
+						afterRollbackProcessorToUse.processBatch(records,
+								Objects.requireNonNullElseGet(recordList, () -> createRecordList(records)),
+								ListenerConsumer.this.consumer,
+								KafkaMessageListenerContainer.this.thisOrParentContainer, e, true,
+								ListenerConsumer.this.eosMode);
 					}
 
 				});
@@ -2195,16 +2199,10 @@ private void batchAfterRollback(final ConsumerRecords<K, V> records,
 				AfterRollbackProcessor<K, V> afterRollbackProcessorToUse) {
 
 			try {
-				if (recordList == null) {
-					afterRollbackProcessorToUse.process(createRecordList(records), this.consumer,
-							KafkaMessageListenerContainer.this.thisOrParentContainer,  rollbackException, false,
-							this.eosMode);
-				}
-				else {
-					afterRollbackProcessorToUse.process(recordList, this.consumer,
-							KafkaMessageListenerContainer.this.thisOrParentContainer,  rollbackException, false,
-							this.eosMode);
-				}
+				afterRollbackProcessorToUse.processBatch(records,
+						Objects.requireNonNullElseGet(recordList, () -> createRecordList(records)), this.consumer,
+						KafkaMessageListenerContainer.this.thisOrParentContainer, rollbackException, true,
+						this.eosMode);
 			}
 			catch (KafkaException ke) {
 				ke.selfLog("AfterRollbackProcessor threw an exception", this.logger);
@@ -2285,9 +2283,7 @@ private void commitOffsetsIfNeeded(final ConsumerRecords<K, V> records) {
 					|| this.producer != null) {
 				if (this.remainingRecords != null) {
 					ConsumerRecord<K, V> firstUncommitted = this.remainingRecords.iterator().next();
-					Iterator<ConsumerRecord<K, V>> it = records.iterator();
-					while (it.hasNext()) {
-						ConsumerRecord<K, V> next = it.next();
+					for (ConsumerRecord<K, V> next : records) {
 						if (!next.equals(firstUncommitted)) {
 							this.acks.add(next);
 						}
@@ -2716,8 +2712,8 @@ private void pauseForNackSleep() {
 		/**
 		 * Actually invoke the listener.
 		 * @param cRecord the record.
-		 * @param iterator the {@link ConsumerRecords} iterator - used only if a
-		 * {@link RemainingRecordsErrorHandler} is being used.
+		 * @param iterator the {@link ConsumerRecords} iterator - only used by
+		 * {@link DefaultErrorHandler#handleRemaining}.
 		 * @return an exception.
 		 * @throws Error an error.
 		 */
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessorTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessorTests.java
@@ -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.
@@ -27,14 +27,18 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
@@ -101,7 +105,6 @@ void testClassifier() {
 
 	@Test
 	void testBatchBackOff() {
-		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
 		@SuppressWarnings("unchecked")
 		KafkaOperations<String, String> template = mock(KafkaOperations.class);
 		given(template.isTransactional()).willReturn(true);
@@ -118,35 +121,64 @@ void testBatchBackOff() {
 		ConsumerRecord<String, String> record1 = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
 		ConsumerRecord<String, String> record2 = new ConsumerRecord<>("foo", 1, 1L, "foo", "bar");
 		List<ConsumerRecord<String, String>> records = Arrays.asList(record1, record2);
+		Map<TopicPartition, List<ConsumerRecord<String, String>>> map = new HashMap<>();
+		records.forEach(rec -> map.computeIfAbsent(new TopicPartition(rec.topic(), rec.partition()),
+				tp -> new ArrayList<>()).add(rec));
+		ConsumerRecords<String, String> consumerRecords = new ConsumerRecords<>(map);
 		IllegalStateException illegalState = new IllegalStateException();
 		@SuppressWarnings("unchecked")
 		Consumer<String, String> consumer = mock(Consumer.class);
 		given(consumer.groupMetadata()).willReturn(new ConsumerGroupMetadata("foo"));
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
 		given(container.isRunning()).willReturn(true);
-		processor.process(records, consumer, container, illegalState, false, EOSMode.V2);
-		processor.process(records, consumer, container, illegalState, false, EOSMode.V2);
+		processor.processBatch(consumerRecords, records, consumer, container, illegalState, true, EOSMode.V2);
+		processor.processBatch(consumerRecords, records, consumer, container, illegalState, true, EOSMode.V2);
 		verify(backOff, times(2)).start();
 		verify(execution.get(), times(2)).nextBackOff();
 		processor.clearThreadState();
-		processor.process(records, consumer, container, illegalState, false, EOSMode.V2);
+		processor.processBatch(consumerRecords, records, consumer, container, illegalState, true, EOSMode.V2);
 		verify(backOff, times(3)).start();
+
 	}
 
-	void testEarlyExitBackOff() {
-		DefaultAfterRollbackProcessor<String, String> processor = new DefaultAfterRollbackProcessor<>(
-				new FixedBackOff(1, 10_000));
+	@Test
+	void testBatchListenerFailedException() {
 		@SuppressWarnings("unchecked")
-		Consumer<String, String> consumer = mock(Consumer.class);
+		KafkaOperations<String, String> template = mock(KafkaOperations.class);
+		given(template.isTransactional()).willReturn(true);
+		BackOff backOff = spy(new FixedBackOff(0, 1));
+		AtomicReference<BackOffExecution> execution = new AtomicReference<>();
+		willAnswer(inv -> {
+			BackOffExecution exec = spy((BackOffExecution) inv.callRealMethod());
+			execution.set(exec);
+			return exec;
+		}).given(backOff).start();
+		ConsumerRecordRecoverer recoverer = mock(ConsumerRecordRecoverer.class);
+		DefaultAfterRollbackProcessor<String, String> processor = new DefaultAfterRollbackProcessor<>(recoverer,
+				backOff, template, false);
 		ConsumerRecord<String, String> record1 = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
+		ConsumerRecord<String, String> record3 = new ConsumerRecord<>("foo", 0, 1L, "foo", "bar");
 		ConsumerRecord<String, String> record2 = new ConsumerRecord<>("foo", 1, 1L, "foo", "bar");
-		List<ConsumerRecord<String, String>> records = Arrays.asList(record1, record2);
-		IllegalStateException illegalState = new IllegalStateException();
+		ConsumerRecord<String, String> record4 = new ConsumerRecord<>("foo", 1, 2L, "foo", "bar");
+		List<ConsumerRecord<String, String>> records = Arrays.asList(record1, record2, record3, record4);
+		Map<TopicPartition, List<ConsumerRecord<String, String>>> map = new HashMap<>();
+		records.forEach(rec -> map.computeIfAbsent(new TopicPartition(rec.topic(), rec.partition()),
+				tp -> new ArrayList<>()).add(rec));
+		ConsumerRecords<String, String> consumerRecords = new ConsumerRecords<>(map);
+		@SuppressWarnings("unchecked")
+		Consumer<String, String> consumer = mock(Consumer.class);
+		given(consumer.groupMetadata()).willReturn(new ConsumerGroupMetadata("foo"));
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
-		given(container.isRunning()).willReturn(false);
-		long t1 = System.currentTimeMillis();
-		processor.process(records, consumer, container, illegalState, true, EOSMode.V2);
-		assertThat(System.currentTimeMillis() < t1 + 5_000);
+		ContainerProperties containerProperties = mock(ContainerProperties.class);
+		given(container.getContainerProperties()).willReturn(containerProperties);
+		given(container.isRunning()).willReturn(true);
+		BatchListenerFailedException batchFailed = new BatchListenerFailedException("", 2);
+		processor.processBatch(consumerRecords, records, consumer, container, batchFailed, true, EOSMode.V2);
+		verify(consumer).seek(new TopicPartition("foo", 1), 1);
+		BatchListenerFailedException batchFailed2 = new BatchListenerFailedException("", record1);
+		processor.processBatch(consumerRecords, records, consumer, container, batchFailed2, true, EOSMode.V2);
+		verify(consumer).seek(new TopicPartition("foo", 0), 0);
+		verify(consumer, times(2)).seek(new TopicPartition("foo", 1), 1);
 	}
 
 	@Test
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

Original file line number	Diff line number	Diff line change
`@@ -98,9 +98,9 @@ public int getIndex() {`
`98`	`98`
`99`	`99`	`@Override`
`100`	`100`	`public String getMessage() {`
`101`		`- return super.getMessage() + (this.record != null`
	`101`	`+ return super.getMessage() + " " + (this.record != null`
`102`	`102`	`? (this.record.topic() + "-" + this.record.partition() + "@" + this.record.offset())`
`103`		`- : (" @-" + this.index));`
	`103`	`+ : ("@-" + this.index));`
`104`	`104`	`}`
`105`	`105`
`106`	`106`	`}`
Original file line number	Diff line number	Diff line change
`@@ -120,7 +120,7 @@ public void setReclassifyOnExceptionChange(boolean reclassifyOnExceptionChange)`
`120`	`120`	`@Override`
`121`	`121`	`protected void notRetryable(Stream<Class<? extends Exception>> notRetryable) {`
`122`	`122`	`if (this.fallbackBatchHandler instanceof ExceptionClassifier handler) {`
`123`		`- notRetryable.forEach(ex -> handler.addNotRetryableExceptions(ex));`
	`123`	`+ notRetryable.forEach(handler::addNotRetryableExceptions);`
`124`	`124`	`}`
`125`	`125`	`}`
`126`	`126`
`@@ -159,11 +159,19 @@ protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> data, C`
`159`	`159`
`160`	`160`	`protected <K, V> ConsumerRecords<K, V> handle(Exception thrownException, ConsumerRecords<?, ?> data,`
`161`	`161`	`Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {`
	`162`	`+ int index = handlerBatchListenerFailedException(thrownException, data);`
	`163`	`+ if (index > -1) {`
	`164`	`+ return seekOrRecover(thrownException, data, consumer, container, index);`
	`165`	`+ }`
	`166`	`+ fallback(thrownException, data, consumer, container, invokeListener);`
	`167`	`+ return ConsumerRecords.empty();`
	`168`	`+ }`
	`169`	`+`
	`170`	`+ protected int handlerBatchListenerFailedException(Exception thrownException, ConsumerRecords<?, ?> data) {`
`162`	`171`
`163`	`172`	`BatchListenerFailedException batchListenerFailedException = getBatchListenerFailedException(thrownException);`
`164`	`173`	`if (batchListenerFailedException == null) {`
`165`	`174`	`this.logger.debug(thrownException, "Expected a BatchListenerFailedException; re-delivering full batch");`
`166`		`- fallback(thrownException, data, consumer, container, invokeListener);`
`167`	`175`	`}`
`168`	`176`	`else {`
`169`	`177`	`getRetryListeners().forEach(listener -> listener.failedDelivery(data, thrownException, 1));`
`@@ -181,13 +189,12 @@ protected <K, V> ConsumerRecords<K, V> handle(Exception thrownException, Consume`
`181`	`189`
`182`	`190`	`}`
`183`	`191`	`});`
`184`		`- fallback(thrownException, data, consumer, container, invokeListener);`
`185`	`192`	`}`
`186`	`193`	`else {`
`187`		`- return seekOrRecover(thrownException, data, consumer, container, index);`
	`194`	`+ return index;`
`188`	`195`	`}`
`189`	`196`	`}`
`190`		`- return ConsumerRecords.empty();`
	`197`	`+ return -1;`
`191`	`198`	`}`
`192`	`199`
`193`	`200`	`private void fallback(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,`
`@@ -201,11 +208,9 @@ private int findIndex(ConsumerRecords<?, ?> data, ConsumerRecord<?, ?> record) {`
`201`	`208`	`return -1;`
`202`	`209`	`}`
`203`	`210`	`int i = 0;`
`204`		`- Iterator<?> iterator = data.iterator();`
`205`		`- while (iterator.hasNext()) {`
`206`		`- ConsumerRecord<?, ?> candidate = (ConsumerRecord<?, ?>) iterator.next();`
`207`		`- if (candidate.topic().equals(record.topic()) && candidate.partition() == record.partition()`
`208`		`- && candidate.offset() == record.offset()) {`
	`211`	`+ for (ConsumerRecord<?, ?> datum : data) {`
	`212`	`+ if (datum.topic().equals(record.topic()) && datum.partition() == record.partition()`
	`213`	`+ && datum.offset() == record.offset()) {`
`209`	`214`	`break;`
`210`	`215`	`}`
`211`	`216`	`i++;`