This is an automated email from the ASF dual-hosted git repository.
sigram pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/main by this push:
new de5e1ecd540 SOLR-18408: CrossDC Consumer incorrect nextOffset commits
(#4831)
de5e1ecd540 is described below
commit de5e1ecd54062e5bbb9491777bbc60bcba20bcdf
Author: Andrzej Białecki <[email protected]>
AuthorDate: Wed Sep 2 10:37:03 2026 +0200
SOLR-18408: CrossDC Consumer incorrect nextOffset commits (#4831)
---
changelog/unreleased/solr-18408.yml | 8 +
.../manager/consumer/KafkaCrossDcConsumer.java | 47 ++--
.../crossdc/manager/consumer/PartitionManager.java | 172 ++++++++++----
.../manager/consumer/KafkaCrossDcConsumerTest.java | 62 ++++-
.../manager/consumer/PartitionManagerTest.java | 251 ++++++++++++++++-----
.../solr/crossdc/common/KafkaMirroringSink.java | 4 +-
.../update/processor/MirroringUpdateProcessor.java | 1 +
7 files changed, 413 insertions(+), 132 deletions(-)
diff --git a/changelog/unreleased/solr-18408.yml
b/changelog/unreleased/solr-18408.yml
new file mode 100644
index 00000000000..85b552c492c
--- /dev/null
+++ b/changelog/unreleased/solr-18408.yml
@@ -0,0 +1,8 @@
+title: SOLR-18408 CrossDC Consumer fix incorrect nextOffset commits
+
+type: fixed
+authors:
+ - name: Andrzej Bialecki
+links:
+ - name: SOLR-18408
+ url: https://issues.apache.org/jira/browse/SOLR-18408
diff --git
a/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumer.java
b/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumer.java
index 055c16fd9f0..b51122646dc 100644
---
a/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumer.java
+++
b/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumer.java
@@ -361,7 +361,7 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
log.trace("Entered pollAndProcessRequests loop");
try {
try {
- partitionManager.checkOffsetUpdates();
+ partitionManager.checkOffsetsAndUpdate();
} catch (Throwable e) {
log.error("Error while checking offset updates, shutting down", e);
return false;
@@ -378,6 +378,9 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
int currentCollapsed = 0;
ConsumerRecord<String, MirroredSolrRequest<?>> lastRecord = null;
+ // the last record merged into updateReqBatch - not the same as
lastRecord once a
+ // subsequent record has triggered a flush of that batch
+ ConsumerRecord<String, MirroredSolrRequest<?>> batchLastRecord = null;
for (TopicPartition partition : records.partitions()) {
if (log.isTraceEnabled()) {
@@ -388,8 +391,6 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
PartitionManager.PartitionWork partitionWork =
partitionManager.getPartitionWork(partition);
PartitionManager.WorkUnit workUnit = new
PartitionManager.WorkUnit(partition);
- workUnit.nextOffset =
PartitionManager.getOffsetForPartition(partitionRecords);
- partitionWork.partitionQueue.add(workUnit);
try {
ModifiableSolrParams lastUpdateParams = null;
for (ConsumerRecord<String, MirroredSolrRequest<?>> requestRecord :
partitionRecords) {
@@ -417,19 +418,6 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
if (log.isTraceEnabled()) {
log.trace("-- picked type={}, params={}", req.getType(), params);
}
- if (topicDebug) {
- solrReq.addHeader("topic.debug", "true");
- solrReq.addHeader("record.topic", requestRecord.topic());
- solrReq.addHeader("record.partition",
String.valueOf(requestRecord.partition()));
- solrReq.addHeader("record.offset",
String.valueOf(requestRecord.offset()));
- solrReq.addHeader("record.timestamp",
String.valueOf(requestRecord.timestamp()));
- solrReq.addHeader("record.key", requestRecord.key());
- solrReq.addHeader("workUnit.nextOffset",
String.valueOf(workUnit.nextOffset));
- solrReq.addHeader("workUnit.partition",
String.valueOf(workUnit.partition));
- solrReq.addHeader("workUnit.topic", workUnit.topic);
- solrReq.addHeader("workUnit.items",
String.valueOf(workUnit.workItems.size()));
- }
-
// determine if it's an UPDATE with deletes, or if the existing
batch has deletes
boolean hasDeletes = false;
if (type == MirroredSolrRequest.Type.UPDATE) {
@@ -456,13 +444,27 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
}
// send previous batch, if any
if (updateReqBatch != null) {
- sendBatch(updateReqBatch, type, lastRecord, workUnit);
+ sendBatch(updateReqBatch, type, batchLastRecord, workUnit);
}
updateReqBatch = null;
currentCollapsed = 0;
workUnit = new PartitionManager.WorkUnit(partition);
- workUnit.nextOffset =
PartitionManager.getOffsetForPartition(partitionRecords);
- partitionWork.partitionQueue.add(workUnit);
+ }
+
+ // this record belongs to the current work unit
+ partitionWork.assignRecord(workUnit, requestRecord.offset());
+
+ if (topicDebug) {
+ solrReq.addHeader("topic.debug", "true");
+ solrReq.addHeader("record.topic", requestRecord.topic());
+ solrReq.addHeader("record.partition",
String.valueOf(requestRecord.partition()));
+ solrReq.addHeader("record.offset",
String.valueOf(requestRecord.offset()));
+ solrReq.addHeader("record.timestamp",
String.valueOf(requestRecord.timestamp()));
+ solrReq.addHeader("record.key", requestRecord.key());
+ solrReq.addHeader("workUnit.nextOffset",
String.valueOf(workUnit.nextOffset));
+ solrReq.addHeader("workUnit.partition",
String.valueOf(workUnit.partition));
+ solrReq.addHeader("workUnit.topic", workUnit.topic);
+ solrReq.addHeader("workUnit.items",
String.valueOf(workUnit.workItems.size()));
}
lastUpdateParams = params;
@@ -481,6 +483,7 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
metrics.incrementCollapsedCounter();
currentCollapsed++;
}
+ batchLastRecord = requestRecord;
UpdateRequest update = (UpdateRequest) solrReq;
MirroredSolrRequest.setParams(updateReqBatch, params);
@@ -510,11 +513,11 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
}
if (updateReqBatch != null) {
- sendBatch(updateReqBatch, MirroredSolrRequest.Type.UPDATE,
lastRecord, workUnit);
+ sendBatch(updateReqBatch, MirroredSolrRequest.Type.UPDATE,
batchLastRecord, workUnit);
updateReqBatch = null;
}
try {
- partitionManager.checkForOffsetUpdates(partition);
+ partitionManager.checkOffsetsAndUpdate(partition);
} catch (Throwable e) {
log.error("Error while checking offset updates, shutting down", e);
return false;
@@ -542,7 +545,7 @@ public class KafkaCrossDcConsumer extends
Consumer.CrossDcConsumer {
}
try {
- partitionManager.checkOffsetUpdates();
+ partitionManager.checkOffsetsAndUpdate();
} catch (Throwable e) {
log.error("Error while checking offset updates, shutting down", e);
return false;
diff --git
a/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/PartitionManager.java
b/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/PartitionManager.java
index cef9bfefb98..5504a5e5b78 100644
---
a/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/PartitionManager.java
+++
b/solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/PartitionManager.java
@@ -20,14 +20,13 @@ import com.google.common.annotations.VisibleForTesting;
import java.lang.invoke.MethodHandles;
import java.util.ArrayDeque;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
@@ -41,8 +40,52 @@ public class PartitionManager {
new ConcurrentHashMap<>();
private final KafkaConsumer<String, MirroredSolrRequest<?>> consumer;
- static class PartitionWork {
+ @VisibleForTesting
+ public static class PartitionWork {
+ final TopicPartition partition;
final Queue<WorkUnit> partitionQueue = new ArrayDeque<>();
+
+ PartitionWork(TopicPartition partition) {
+ this.partition = partition;
+ }
+
+ /**
+ * Assign a record to a work unit: enqueue the unit on its first record,
and advance its commit
+ * point to just past that record. A unit that never receives a record is
never enqueued, so it
+ * can never commit an offset of its own.
+ *
+ * <p>Guarded by the same monitor as {@link
+ * PartitionManager#checkOffsetsAndUpdate(TopicPartition)}, which is the
other place the queue
+ * is touched.
+ *
+ * @param unit the work unit the record belongs to
+ * @param recordOffset offset of the record being assigned
+ * @throws IllegalStateException if recordOffset regresses behind the last
record already
+ * assigned to this unit
+ */
+ synchronized void assignRecord(WorkUnit unit, long recordOffset) {
+ // does this work unit belong to the partition we're interested in?
+ if (unit.partition != partition.partition()) {
+ throw new IllegalStateException(
+ "Work unit for partition "
+ + partition.partition()
+ + " but record for partition "
+ + unit.partition);
+ }
+ if (recordOffset < unit.nextOffset) {
+ throw new IllegalStateException(
+ "Out-of-order record offset "
+ + recordOffset
+ + ", expected an offset greater than or equal to "
+ + (unit.nextOffset - 1));
+ }
+ // if this is a new unit enqueue it first
+ if (unit.nextOffset < 0) {
+ partitionQueue.add(unit);
+ }
+ // advance the commit point to just past the record
+ unit.nextOffset = recordOffset + 1;
+ }
}
@VisibleForTesting
@@ -50,7 +93,12 @@ public class PartitionManager {
final int partition;
final String topic;
final Set<Future<?>> workItems = new HashSet<>();
- long nextOffset;
+
+ /**
+ * Exclusive upper bound of the offsets this unit owns, i.e. the offset to
commit once all of
+ * its work items are done. Negative until the unit is assigned its first
record.
+ */
+ long nextOffset = -1;
WorkUnit(TopicPartition partition) {
this.partition = partition.partition();
@@ -67,56 +115,101 @@ public class PartitionManager {
partition,
(k, v) -> {
if (v == null) {
- return new PartitionWork();
+ return new PartitionWork(partition);
}
return v;
});
}
- public void checkOffsetUpdates() throws Throwable {
+ public void checkOffsetsAndUpdate() throws Throwable {
for (TopicPartition partition : partitionWorkMap.keySet()) {
- checkForOffsetUpdates(partition);
+ checkOffsetsAndUpdate(partition);
}
}
- void checkForOffsetUpdates(TopicPartition partition) throws Throwable {
- synchronized (partition) {
- PartitionWork work;
- if ((work = partitionWorkMap.get(partition)) != null) {
- WorkUnit workUnit = work.partitionQueue.peek();
- if (workUnit != null) {
- boolean allFuturesDone = true;
- for (Future<?> future : workUnit.workItems) {
- if (!future.isDone()) {
- if (log.isTraceEnabled()) {
- log.trace("Future for update is not done topic={}",
partition.topic());
- }
- allFuturesDone = false;
- break;
- }
+ void checkOffsetsAndUpdate(TopicPartition partition) throws Throwable {
+ // can't synchronize on the argument (equal but distinct object for
different threads)
+ // sync on the PartitionWork instead, which is unique per partition and
shared by all threads
+ // that work on that partition.
+ final PartitionWork partitionWork = partitionWorkMap.get(partition);
+ // normally impossible because consumer should always call
#getPartitionWork first
+ // which creates the instance if it doesn't exist.
+ if (partitionWork == null) {
+ throw new IllegalStateException(
+ "PartitionWork for partition " + partition + " not found, likely
programming error.");
+ }
- try {
- future.get();
- } catch (InterruptedException e) {
- log.error("Error updating offset for partition: {}", partition,
e);
- throw e;
- } catch (ExecutionException e) {
- log.error("Error updating offset for partition: {}", partition,
e);
- throw e.getCause();
+ synchronized (partitionWork) {
+ // remove every completed work unit at the head of the queue, stopping
at the first one
+ // that is still in flight - a work unit's offset may only be committed
once all of the
+ // work units before it have been committed too.
+ long committableOffset = -1;
+ WorkUnit workUnit;
+ Throwable failure = null;
+ try {
+ while ((workUnit = partitionWork.partitionQueue.peek()) != null) {
+ if (!isComplete(workUnit, partition)) {
+ break;
+ }
+ // remove completed unit
+ partitionWork.partitionQueue.poll();
+ committableOffset = workUnit.nextOffset;
+ }
+ } catch (Throwable t) {
+ failure = t;
+ throw t;
+ } finally {
+ // commit whatever progress was already verified in this drain, even
if a later
+ // unit's isComplete() threw - otherwise that progress silently gets
lost.
+ if (committableOffset >= 0) {
+ try {
+ updateOffset(partition, committableOffset);
+ } catch (Throwable commitFailure) {
+ if (commitFailure instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
}
-
- if (log.isTraceEnabled()) {
- log.trace("Future for update is done topic={}",
partition.topic());
+ // don't let a secondary commit failure mask the real work-item
failure
+ if (failure != null) {
+ failure.addSuppressed(commitFailure);
+ } else {
+ throw commitFailure;
}
}
+ }
+ }
+ }
+ }
- if (allFuturesDone) {
- work.partitionQueue.poll();
- updateOffset(partition, workUnit.nextOffset);
- }
+ /** Check whether all the work items of this unit are done, rethrowing any
of their failures. */
+ private boolean isComplete(WorkUnit workUnit, TopicPartition partition)
throws Throwable {
+ for (Future<?> future : workUnit.workItems) {
+ if (!future.isDone()) {
+ if (log.isTraceEnabled()) {
+ log.trace("Future for update is not done topic={}",
partition.topic());
}
+ return false;
+ }
+
+ try {
+ // the future is already done, so this returns (or rethrows) without
waiting
+ future.get();
+ } catch (InterruptedException e) {
+ log.error("Error updating offset for partition (interrupted): {}",
partition, e);
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (CancellationException e) {
+ log.error("Error updating offset for partition (cancelled): {}",
partition, e);
+ throw e;
+ } catch (ExecutionException e) {
+ log.error("Error updating offset for partition: {}", partition, e);
+ throw e.getCause();
+ }
+
+ if (log.isTraceEnabled()) {
+ log.trace("Future for update is done topic={}", partition.topic());
}
}
+ return true;
}
/**
@@ -136,9 +229,4 @@ public class PartitionManager {
consumer.commitSync(Map.of(partition, new OffsetAndMetadata(nextOffset)));
}
-
- static long getOffsetForPartition(
- List<ConsumerRecord<String, MirroredSolrRequest<?>>> partitionRecords) {
- return partitionRecords.get(partitionRecords.size() - 1).offset() + 1;
- }
}
diff --git
a/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumerTest.java
b/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumerTest.java
index 2bc50db48fd..3ce4a2b42d3 100644
---
a/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumerTest.java
+++
b/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/KafkaCrossDcConsumerTest.java
@@ -229,7 +229,6 @@ public class KafkaCrossDcConsumerTest {
@Test
public void testHandleFailedResubmit() throws Exception {
- // Set up the KafkaCrossDcConsumer
KafkaConsumer<String, MirroredSolrRequest<?>> mockConsumer =
mock(KafkaConsumer.class);
KafkaCrossDcConsumer consumer = createCrossDcConsumerSpy(mockConsumer);
@@ -245,9 +244,6 @@ public class KafkaCrossDcConsumerTest {
MirroredSolrRequest<?> request = new MirroredSolrRequest<>(new
UpdateRequest());
IQueueHandler.Result<MirroredSolrRequest<?>> failedResubmitResult =
new IQueueHandler.Result<>(IQueueHandler.ResultStatus.FAILED_RESUBMIT,
null, request);
- // SolrMessageProcessor mockMessageProcessor =
mock(SolrMessageProcessor.class);
- // when(mockMessageProcessor.handleItem(any(MirroredSolrRequest.class)))
- // .thenReturn(failedResubmitResult);
// Mock the KafkaMirroringSink
KafkaMirroringSink mockKafkaMirroringSink = mock(KafkaMirroringSink.class);
@@ -406,7 +402,7 @@ public class KafkaCrossDcConsumerTest {
}
// Create a valid MirroredSolrRequest
ConsumerRecord<String, MirroredSolrRequest<?>> record =
- new ConsumerRecord<>("test-topic", 0, 0, "key", new
MirroredSolrRequest<>(validRequest));
+ new ConsumerRecord<>("test-topic", 0, i, "key", new
MirroredSolrRequest<>(validRequest));
records.add(record);
}
ConsumerRecords<String, MirroredSolrRequest<?>> consumerRecords =
@@ -421,6 +417,62 @@ public class KafkaCrossDcConsumerTest {
.sendBatch(any(), eq(MirroredSolrRequest.Type.UPDATE), any(), any());
}
+ /**
+ * When a record's differing params force a flush of the batch collapsed so
far, the flush must be
+ * attributed to the last record actually merged into that batch, not to the
record that merely
+ * triggered the flush.
+ */
+ @Test
+ public void testFlushedBatchLastRecord() {
+ KafkaConsumer<String, MirroredSolrRequest<?>> mockConsumer =
mock(KafkaConsumer.class);
+ KafkaCrossDcConsumer spyConsumer = createCrossDcConsumerSpy(mockConsumer);
+ doReturn(new IQueueHandler.Result<>(IQueueHandler.ResultStatus.HANDLED,
null))
+ .when(messageProcessorMock)
+ .handleItem(any());
+
+ UpdateRequest batchRequest1 = new UpdateRequest();
+ SolrInputDocument doc1 = new SolrInputDocument();
+ doc1.addField("id", "1");
+ batchRequest1.add(doc1);
+
+ UpdateRequest batchRequest2 = new UpdateRequest();
+ SolrInputDocument doc2 = new SolrInputDocument();
+ doc2.addField("id", "2");
+ batchRequest2.add(doc2);
+
+ // different params from the first two records, so it can't collapse with
them and instead
+ // forces a flush of the batch they collapsed into
+ UpdateRequest differentParamsRequest = new UpdateRequest();
+ SolrInputDocument doc3 = new SolrInputDocument();
+ doc3.addField("id", "3");
+ differentParamsRequest.add(doc3);
+ differentParamsRequest.getParams().set("some.param", "different");
+
+ ConsumerRecord<String, MirroredSolrRequest<?>> record1 =
+ new ConsumerRecord<>("test-topic", 0, 0, "key1", new
MirroredSolrRequest<>(batchRequest1));
+ ConsumerRecord<String, MirroredSolrRequest<?>> record2 =
+ new ConsumerRecord<>("test-topic", 0, 1, "key2", new
MirroredSolrRequest<>(batchRequest2));
+ ConsumerRecord<String, MirroredSolrRequest<?>> record3 =
+ new ConsumerRecord<>(
+ "test-topic", 0, 2, "key3", new
MirroredSolrRequest<>(differentParamsRequest));
+
+ ConsumerRecords<String, MirroredSolrRequest<?>> records =
+ new ConsumerRecords<>(
+ Map.of(new TopicPartition("test-topic", 0), List.of(record1,
record2, record3)));
+
+ when(mockConsumer.poll(any())).thenReturn(records).thenThrow(new
WakeupException());
+
+ spyConsumer.run();
+
+ // record1 and record2 collapsed into one batch; that batch's flush must
be attributed to
+ // record2 (its last record), never to record3 (which only triggered the
flush)
+ verify(spyConsumer, times(1))
+ .sendBatch(any(), eq(MirroredSolrRequest.Type.UPDATE), eq(record2),
any());
+ // record3 starts (and, at end of loop, flushes) its own batch
+ verify(spyConsumer, times(1))
+ .sendBatch(any(), eq(MirroredSolrRequest.Type.UPDATE), eq(record3),
any());
+ }
+
@Test
public void testHandleInvalidMirroredSolrRequest() {
KafkaConsumer<String, MirroredSolrRequest<?>> mockConsumer =
mock(KafkaConsumer.class);
diff --git
a/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/PartitionManagerTest.java
b/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/PartitionManagerTest.java
index 91902f5f896..176aa16d276 100644
---
a/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/PartitionManagerTest.java
+++
b/solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/PartitionManagerTest.java
@@ -19,13 +19,20 @@ package org.apache.solr.crossdc.manager.consumer;
import static org.apache.solr.SolrTestCaseJ4.assumeWorkingMockito;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -35,9 +42,14 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.SolrNamedThreadFactory;
import org.apache.solr.crossdc.common.MirroredSolrRequest;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+/**
+ * Tests that a partition's commit point never runs ahead of work that is
still in flight: a work
+ * unit's offset may only be committed once that unit and every unit queued
before it are done.
+ */
@SuppressWarnings("unchecked")
public class PartitionManagerTest {
@@ -46,15 +58,150 @@ public class PartitionManagerTest {
assumeWorkingMockito();
}
+ private static final TopicPartition PARTITION = new TopicPartition("topic1",
0);
+
+ @SuppressWarnings("unchecked")
+ private final KafkaConsumer<String, MirroredSolrRequest<?>> consumer =
mock(KafkaConsumer.class);
+
+ private PartitionManager partitionManager;
+ private PartitionManager.PartitionWork work;
+
+ @Before
+ public void setUp() {
+ partitionManager = new PartitionManager(consumer);
+ work = partitionManager.getPartitionWork(PARTITION);
+ }
+
+ /** Enqueue a work unit owning a single record at the given offset. */
+ private PartitionManager.WorkUnit enqueue(long recordOffset) {
+ PartitionManager.WorkUnit workUnit = new
PartitionManager.WorkUnit(PARTITION);
+ work.assignRecord(workUnit, recordOffset);
+ return workUnit;
+ }
+
+ @Test
+ public void testDrainsAllCompletedUnitsInSingleCommit() throws Throwable {
+ PartitionManager.WorkUnit first = enqueue(109);
+ PartitionManager.WorkUnit second = enqueue(119);
+ PartitionManager.WorkUnit third = enqueue(129);
+
+ // the later units finish first - nothing may be committed while the head
is in flight
+ CompletableFuture<Void> firstWork = new CompletableFuture<>();
+ first.workItems.add(firstWork);
+ second.workItems.add(CompletableFuture.completedFuture(null));
+ third.workItems.add(CompletableFuture.completedFuture(null));
+
+ partitionManager.checkOffsetsAndUpdate(PARTITION);
+
+ verify(consumer, never()).commitSync(anyMap());
+ assertEquals(3, work.partitionQueue.size());
+
+ // once the head completes, all three retire under a single commit of the
furthest offset
+ firstWork.complete(null);
+ partitionManager.checkOffsetsAndUpdate(PARTITION);
+
+ verify(consumer).commitSync(Map.of(PARTITION, new OffsetAndMetadata(130)));
+ verifyNoMoreInteractions(consumer);
+ assertEquals(0, work.partitionQueue.size());
+ }
+
+ @Test
+ public void testStopsAtFirstIncompleteUnit() throws Throwable {
+ PartitionManager.WorkUnit first = enqueue(109);
+ PartitionManager.WorkUnit second = enqueue(119);
+ enqueue(129);
+
+ first.workItems.add(CompletableFuture.completedFuture(null));
+ second.workItems.add(new CompletableFuture<>());
+
+ partitionManager.checkOffsetsAndUpdate(PARTITION);
+
+ // only the first unit's records are done, so only its offset may be
committed
+ verify(consumer).commitSync(Map.of(PARTITION, new OffsetAndMetadata(110)));
+ verifyNoMoreInteractions(consumer);
+ assertEquals(2, work.partitionQueue.size());
+ assertSame(second, work.partitionQueue.peek());
+ }
+
+ @Test
+ public void testAssignRecordThrowsOnOutOfOrderOffset() {
+ PartitionManager.WorkUnit unit = new PartitionManager.WorkUnit(PARTITION);
+ work.assignRecord(unit, 109);
+
+ try {
+ work.assignRecord(unit, 108);
+ fail("expected an out-of-order record offset to be rejected");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void testFailedWorkItemPropagatesAndBlocksTheCommit() {
+ PartitionManager.WorkUnit first = enqueue(109);
+ first.workItems.add(CompletableFuture.failedFuture(new
IllegalStateException("boom")));
+
+ Throwable thrown = null;
+ try {
+ partitionManager.checkOffsetsAndUpdate(PARTITION);
+ } catch (Throwable e) {
+ thrown = e;
+ assertEquals(IllegalStateException.class, e.getClass());
+ assertEquals("boom", e.getMessage());
+ }
+ if (thrown == null) {
+ fail("expected the work item failure to be rethrown");
+ }
+
+ verify(consumer, never()).commitSync(anyMap());
+ }
+
+ /**
+ * When a later unit's isComplete() fails after an earlier unit already
requires a commit, and the
+ * commit itself then also fails (e.g. a partition rebalance mid-drain), the
original work-item
+ * failure must still be the one that propagates - with the commit failure
attached as suppressed
+ * rather than replacing the original failure.
+ */
+ @Test
+ public void testCommitFailurePropgates() {
+ PartitionManager.WorkUnit first = enqueue(109);
+ PartitionManager.WorkUnit second = enqueue(119);
+
+ first.workItems.add(CompletableFuture.completedFuture(null));
+ second.workItems.add(CompletableFuture.failedFuture(new
IllegalStateException("boom")));
+
+ RuntimeException commitFailure = new RuntimeException("commit failed");
+ doThrow(commitFailure).when(consumer).commitSync(anyMap());
+
+ Throwable thrown = null;
+ try {
+ partitionManager.checkOffsetsAndUpdate(PARTITION);
+ } catch (Throwable e) {
+ thrown = e;
+ // the real root cause must still be the one that surfaces
+ assertEquals(IllegalStateException.class, thrown.getClass());
+ assertEquals("boom", thrown.getMessage());
+ // with the secondary commit failure preserved rather than discarded
+ assertEquals(1, thrown.getSuppressed().length);
+ assertSame(commitFailure, thrown.getSuppressed()[0]);
+ }
+ if (thrown == null) {
+ fail("expected the work item failure to be rethrown");
+ }
+
+ // the earlier unit's offset was still attempted, even though the commit
itself failed
+ verify(consumer).commitSync(Map.of(PARTITION, new OffsetAndMetadata(110)));
+ assertEquals(1, work.partitionQueue.size());
+ assertSame(second, work.partitionQueue.peek());
+ }
+
/**
* Should return the existing PartitionWork when the partition is already in
the partitionWorkMap
*/
@Test
- public void getPartitionWorkWhenPartitionInMap() {
- KafkaConsumer<String, MirroredSolrRequest<?>> consumer =
mock(KafkaConsumer.class);
- PartitionManager partitionManager = new PartitionManager(consumer);
+ public void testPartitionWorkWhenPartitionInMap() {
TopicPartition partition = new TopicPartition("test-topic", 0);
- PartitionManager.PartitionWork partitionWork = new
PartitionManager.PartitionWork();
+ PartitionManager.PartitionWork partitionWork = new
PartitionManager.PartitionWork(partition);
partitionManager.partitionWorkMap.put(partition, partitionWork);
PartitionManager.PartitionWork result =
partitionManager.getPartitionWork(partition);
@@ -65,9 +212,7 @@ public class PartitionManagerTest {
/** Should create a new PartitionWork when the partition is not in the
partitionWorkMap */
@Test
- public void getPartitionWorkWhenPartitionNotInMap() {
- KafkaConsumer<String, MirroredSolrRequest<?>> consumer =
mock(KafkaConsumer.class);
- PartitionManager partitionManager = new PartitionManager(consumer);
+ public void testPartitionWorkWhenPartitionNotInMap() {
TopicPartition partition = new TopicPartition("test-topic", 0);
PartitionManager.PartitionWork partitionWork =
partitionManager.getPartitionWork(partition);
@@ -79,83 +224,65 @@ public class PartitionManagerTest {
/** Should not update the offset when the future for update is not done */
@Test
- public void checkForOffsetUpdatesWhenFutureNotDone() throws Throwable {
- KafkaConsumer<String, MirroredSolrRequest<?>> consumer =
mock(KafkaConsumer.class);
- PartitionManager partitionManager = new PartitionManager(consumer);
- TopicPartition partition = new TopicPartition("test-topic", 0);
- PartitionManager.PartitionWork partitionWork =
partitionManager.getPartitionWork(partition);
- PartitionManager.WorkUnit workUnit = new
PartitionManager.WorkUnit(partition);
+ public void testForOffsetUpdatesWhenFutureNotDone() throws Throwable {
+ PartitionManager.WorkUnit workUnit = new
PartitionManager.WorkUnit(PARTITION);
Future<?> future = mock(Future.class);
when(future.isDone()).thenReturn(false);
workUnit.workItems.add(future);
- partitionWork.partitionQueue.add(workUnit);
+ work.assignRecord(workUnit, 0);
- partitionManager.checkForOffsetUpdates(partition);
+ partitionManager.checkOffsetsAndUpdate(PARTITION);
- assertEquals(1, partitionWork.partitionQueue.size());
- assertTrue(partitionWork.partitionQueue.contains(workUnit));
+ assertEquals(1, work.partitionQueue.size());
+ assertTrue(work.partitionQueue.contains(workUnit));
}
/** Should update the offset when the future for update is done */
@Test
- public void checkForOffsetUpdatesWhenFutureDone() throws Throwable {
- KafkaConsumer<String, MirroredSolrRequest<?>> consumer =
mock(KafkaConsumer.class);
- PartitionManager partitionManager = new PartitionManager(consumer);
- TopicPartition partition = new TopicPartition("test-topic", 0);
-
- PartitionManager.PartitionWork partitionWork =
partitionManager.getPartitionWork(partition);
- PartitionManager.WorkUnit workUnit = new
PartitionManager.WorkUnit(partition);
- partitionWork.partitionQueue.add(workUnit);
+ public void testForOffsetUpdatesWhenFutureDone() throws Throwable {
+ PartitionManager.WorkUnit workUnit = new
PartitionManager.WorkUnit(PARTITION);
+ work.assignRecord(workUnit, 0);
// Use a real Future instead of a mocked one
ExecutorService executor =
ExecutorUtil.newMDCAwareSingleThreadExecutor(new
SolrNamedThreadFactory("test"));
- Future<?> future =
- executor.submit(
- () -> {
- // Simulate the task being completed
- });
+ try {
+ Future<?> future =
+ executor.submit(
+ () -> {
+ // Simulate the task being completed
+ });
- workUnit.workItems.add(future);
-
- // Wait for the Future to completeE
- future.get(10, TimeUnit.SECONDS);
+ workUnit.workItems.add(future);
- partitionManager.checkForOffsetUpdates(partition);
+ // Wait for the Future to completeE
+ future.get(10, TimeUnit.SECONDS);
- // Verify that the consumer.commitSync() method was called with the
correct parameters
- verify(consumer, times(1))
- .commitSync(Map.of(partition, new
OffsetAndMetadata(workUnit.nextOffset)));
+ partitionManager.checkOffsetsAndUpdate(PARTITION);
- // Verify that the partitionQueue is empty after processing
- assertTrue(partitionWork.partitionQueue.isEmpty());
+ // Verify that the consumer.commitSync() method was called with the
correct parameters
+ verify(consumer, times(1))
+ .commitSync(Map.of(PARTITION, new
OffsetAndMetadata(workUnit.nextOffset)));
- // Shutdown the executor
- executor.shutdown();
+ // Verify that the partitionQueue is empty after processing
+ assertTrue(work.partitionQueue.isEmpty());
+ } finally {
+ executor.shutdown();
+ }
}
/** Should check for offset updates for all partitions in the
partitionWorkMap */
@Test
- public void checkOffsetUpdatesForAllPartitions() throws Throwable { //
Create a mock KafkaConsumer
- KafkaConsumer<String, MirroredSolrRequest<?>> mockConsumer =
mock(KafkaConsumer.class);
-
- // Create a PartitionManager instance with the mock KafkaConsumer
- PartitionManager partitionManager = new PartitionManager(mockConsumer);
-
- // Create a few TopicPartitions
- TopicPartition partition1 = new TopicPartition("topic1", 0);
+ public void testOffsetUpdatesForAllPartitions() throws Throwable {
+ // reuse the shared partition for the first partition, and add a second one
TopicPartition partition2 = new TopicPartition("topic2", 0);
-
- // Add some PartitionWork to the partitionWorkMap
- PartitionManager.PartitionWork work1 =
partitionManager.getPartitionWork(partition1);
PartitionManager.PartitionWork work2 =
partitionManager.getPartitionWork(partition2);
- // Create WorkUnits and add them to the PartitionWork
- PartitionManager.WorkUnit workUnit1 = new
PartitionManager.WorkUnit(partition1);
+ PartitionManager.WorkUnit workUnit1 = new
PartitionManager.WorkUnit(PARTITION);
PartitionManager.WorkUnit workUnit2 = new
PartitionManager.WorkUnit(partition2);
- work1.partitionQueue.add(workUnit1);
- work2.partitionQueue.add(workUnit2);
+ work.assignRecord(workUnit1, 0);
+ work2.assignRecord(workUnit2, 0);
// Create mock Futures and add them to the WorkUnits
Future<?> mockFuture1 = mock(Future.class);
@@ -168,17 +295,17 @@ public class PartitionManagerTest {
when(mockFuture1.isDone()).thenReturn(true);
when(mockFuture2.isDone()).thenReturn(true);
- // Call the checkOffsetUpdates method
- partitionManager.checkOffsetUpdates();
+ // Call the checkOffsetsAndUpdate method
+ partitionManager.checkOffsetsAndUpdate();
// Verify that the futures were checked for completion
verify(mockFuture1, times(1)).isDone();
verify(mockFuture2, times(1)).isDone();
// Verify that the updateOffset method was called for each partition
- verify(mockConsumer, times(1))
- .commitSync(Map.of(partition1, new
OffsetAndMetadata(workUnit1.nextOffset)));
- verify(mockConsumer, times(1))
+ verify(consumer, times(1))
+ .commitSync(Map.of(PARTITION, new
OffsetAndMetadata(workUnit1.nextOffset)));
+ verify(consumer, times(1))
.commitSync(Map.of(partition2, new
OffsetAndMetadata(workUnit2.nextOffset)));
}
}
diff --git
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
index c348f032c21..8935160937f 100644
---
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
+++
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
@@ -34,6 +34,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.solr.common.util.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -118,7 +119,7 @@ public class KafkaMirroringSink implements
RequestMirroringSink, Closeable {
slowSubmitAction(elapsedTimeMillis);
}
} catch (Exception e) {
- // We are intentionally catching all exceptions, the expected exception
form this function is
+ // We are intentionally catching all exceptions, the expected exception
from this function is
// {@link MirroringException}
String message =
"Unable to enqueue request "
@@ -232,5 +233,6 @@ public class KafkaMirroringSink implements
RequestMirroringSink, Closeable {
producer.flush();
producer.close();
}
+ IOUtils.closeQuietly(consumer);
}
}
diff --git
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateProcessor.java
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateProcessor.java
index 284dd91ca7d..223616806d6 100644
---
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateProcessor.java
+++
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/update/processor/MirroringUpdateProcessor.java
@@ -273,6 +273,7 @@ public class MirroringUpdateProcessor extends
UpdateRequestProcessor {
} catch (Exception e) {
log.error("mirror submit failed", e);
producerMetrics.getSubmittedDeleteByIdError().inc();
+ producerMetrics.getSubmitError().inc();
throw new SolrException(SERVER_ERROR, "mirror submit failed", e);
}
}
