sigram commented on code in PR #4831:
URL: https://github.com/apache/solr/pull/4831#discussion_r3902513146
##########
solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/consumer/PartitionManagerTest.java:
##########
@@ -46,11 +57,109 @@ public static void ensureWorkingMockito() {
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());
+ }
+
/**
* Should return the existing PartitionWork when the partition is already in
the partitionWorkMap
*/
@Test
- public void getPartitionWorkWhenPartitionInMap() {
+ public void testPartitionWorkWhenPartitionInMap() {
Review Comment:
It's a test method not a getter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
