This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new fe6511365d0 Simplify re-ingestion tracking and skip obsolete queued
jobs (#19893)
fe6511365d0 is described below
commit fe6511365d0e067b15edf96e13b529c57fd14d30
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Oct 16 12:22:12 2026 -0700
Simplify re-ingestion tracking and skip obsolete queued jobs (#19893)
---
.../server/api/resources/ReingestionResource.java | 71 +++++++------
.../ReingestionConsumptionTimeoutTest.java | 13 +--
.../api/resources/ReingestionResourceApiTest.java | 112 ++++++++++++++++++---
3 files changed, 140 insertions(+), 56 deletions(-)
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java
index 6264f4c4156..e6256156c80 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java
@@ -40,10 +40,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BooleanSupplier;
-import javax.annotation.Nullable;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
import javax.inject.Singleton;
@@ -66,6 +63,7 @@ import
org.apache.pinot.segment.local.realtime.writer.StatelessRealtimeSegmentWr
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.server.realtime.ServerSegmentCompletionProtocolHandler;
import org.apache.pinot.server.starter.ServerInstance;
+import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -93,14 +91,11 @@ public class ReingestionResource {
static final int MAX_PARALLEL_REINGESTIONS =
Math.max(Runtime.getRuntime().availableProcessors() / 2, 1);
public static final long CHECK_INTERVAL_MS =
Duration.ofSeconds(5).toMillis();
- // Tracks if a particular segment is currently being re-ingested
- private final ConcurrentHashMap<String, AtomicBoolean> _reingestingSegments
= new ConcurrentHashMap<>();
-
// Executor for asynchronous re-ingestion
private final ExecutorService _reingestionExecutor =
Executors.newFixedThreadPool(MAX_PARALLEL_REINGESTIONS,
new
ThreadFactoryBuilder().setNameFormat("reingestion-worker-%d").setDaemon(true).build());
- // Keep track of jobs by jobId => job info
+ // Tracks queued and running jobs by segment name, also preventing duplicate
repairs
private final ConcurrentHashMap<String, ReingestionJob> _runningJobs = new
ConcurrentHashMap<>();
@Inject
@@ -142,7 +137,6 @@ public class ReingestionResource {
@ApiOperation("Get all running re-ingestion jobs, including the ones waiting
for a re-ingestion thread, along with "
+ "job IDs")
public Response getAllRunningReingestionJobs() {
- // Filter only the jobs still marked as running, including the ones
waiting for a re-ingestion thread
List<ReingestionJob> runningJobs = new ArrayList<>(_runningJobs.values());
return Response.ok(runningJobs).build();
}
@@ -198,42 +192,31 @@ public class ReingestionResource {
Response.Status.INTERNAL_SERVER_ERROR);
}
- IndexLoadingConfig indexLoadingConfig =
tableDataManager.fetchIndexLoadingConfig();
-
- // Check if this segment is already being re-ingested
- AtomicBoolean isIngesting =
_reingestingSegments.computeIfAbsent(segmentName, k -> new
AtomicBoolean(false));
- if (!isIngesting.compareAndSet(false, true)) {
+ String jobId = UUID.randomUUID().toString();
+ ReingestionJob job = new ReingestionJob(jobId, segmentName);
+ // Track the job before submitting it, so queued jobs are listed and
duplicate requests are rejected
+ if (_runningJobs.putIfAbsent(segmentName, job) != null) {
return Response.status(Response.Status.CONFLICT)
.entity("Re-ingestion for segment: " + segmentName + " is already in
progress.")
.build();
}
- // Generate a jobId for tracking
- String jobId = UUID.randomUUID().toString();
- ReingestionJob job = new ReingestionJob(jobId, segmentName);
-
- // Track the job before submitting it, so that it is listed while waiting
for a re-ingestion thread
- _runningJobs.put(jobId, job);
-
// Kick off the actual work asynchronously
try {
_reingestionExecutor.submit(() -> {
try {
- doReingestSegment(realtimeTableName, segmentZKMetadata,
indexLoadingConfig,
- tableDataManager.getSegmentBuildSemaphore());
+ doReingestSegment(realtimeTableName, segmentName);
} catch (Exception e) {
LOGGER.error("Error during async re-ingestion for job {}
(segment={})", jobId, segmentName, e);
_serverInstance.getServerMetrics()
.addMeteredTableValue(realtimeTableName,
ServerMeter.SEGMENT_REINGESTION_FAILURE, 1);
} finally {
- _runningJobs.remove(jobId);
- _reingestingSegments.remove(segmentName);
+ _runningJobs.remove(segmentName, job);
}
});
} catch (RejectedExecutionException e) {
// The executor is shut down when the admin API stops
- _runningJobs.remove(jobId);
- _reingestingSegments.remove(segmentName);
+ _runningJobs.remove(segmentName, job);
throw new WebApplicationException("Server is shutting down, cannot
re-ingest segment: " + segmentName,
Response.Status.SERVICE_UNAVAILABLE);
}
@@ -249,14 +232,38 @@ public class ReingestionResource {
LOGGER.info("Stopped re-ingestion executor, dropped {} jobs waiting for a
re-ingestion thread", numDroppedJobs);
}
- /// The actual re-ingestion logic, moved into a separate method for clarity.
- /// This is essentially the old synchronous logic you had in reingestSegment.
- private void doReingestSegment(String realtimeTableName, SegmentZKMetadata
segmentZKMetadata,
- IndexLoadingConfig indexLoadingConfig, @Nullable Semaphore
segmentBuildSemaphore)
+ private void doReingestSegment(String realtimeTableName, String segmentName)
throws Exception {
- String segmentName = segmentZKMetadata.getSegmentName();
+ // The table or segment may have been removed or repaired on another
replica while this job was queued.
+ RealtimeTableDataManager tableDataManager =
+ (RealtimeTableDataManager)
_serverInstance.getInstanceDataManager().getTableDataManager(realtimeTableName);
+ if (tableDataManager == null) {
+ LOGGER.warn("Skipping re-ingestion of segment: {} because table: {} is
no longer available", segmentName,
+ realtimeTableName);
+ return;
+ }
+ if (tableDataManager.isShutDown()) {
+ LOGGER.warn("Skipping re-ingestion of segment: {} because table: {} is
shut down", segmentName,
+ realtimeTableName);
+ return;
+ }
+ SegmentZKMetadata segmentZKMetadata =
tableDataManager.fetchZKMetadataNullable(segmentName);
+ if (segmentZKMetadata == null) {
+ LOGGER.warn("Skipping re-ingestion of segment: {} because its metadata
no longer exists", segmentName);
+ return;
+ }
+ if (segmentZKMetadata.getStatus() != Status.COMMITTING) {
+ LOGGER.warn("Skipping re-ingestion of segment: {} because its status is
{} (expected COMMITTING)", segmentName,
+ segmentZKMetadata.getStatus());
+ return;
+ }
+ if (segmentZKMetadata.getDownloadUrl() != null) {
+ LOGGER.warn("Skipping re-ingestion of segment: {} because its download
URL is already present", segmentName);
+ return;
+ }
+ IndexLoadingConfig indexLoadingConfig =
tableDataManager.fetchIndexLoadingConfig();
try (StatelessRealtimeSegmentWriter writer = new
StatelessRealtimeSegmentWriter(segmentZKMetadata,
- indexLoadingConfig, segmentBuildSemaphore)) {
+ indexLoadingConfig, tableDataManager.getSegmentBuildSemaphore())) {
// Read when the consumption starts, so that a job waiting for a
re-ingestion thread uses the latest timeout
long consumptionTimeoutMs = _consumptionTimeout.getTimeoutMs();
LOGGER.info("Starting consumption to re-ingest segment: {} with timeout:
{}ms", segmentName,
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeoutTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeoutTest.java
index 2c04697c02f..83e67f1e76f 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeoutTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeoutTest.java
@@ -44,18 +44,6 @@ public class ReingestionConsumptionTimeoutTest {
assertThat(createTimeout(EMPTY_SERVER_CONF,
Map.of()).getTimeoutMs()).isEqualTo(DEFAULT_TIMEOUT_MS);
}
- @Test
- public void testServerConfig() {
- assertThat(createTimeout(SERVER_CONF,
Map.of()).getTimeoutMs()).isEqualTo(3_600_000L);
- }
-
- @Test
- public void testClusterConfigOverridesServerConfig() {
- Map<String, String> clusterConfigs =
Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, "7200000");
- assertThat(createTimeout(SERVER_CONF,
clusterConfigs).getTimeoutMs()).isEqualTo(7_200_000L);
- assertThat(createTimeout(EMPTY_SERVER_CONF,
clusterConfigs).getTimeoutMs()).isEqualTo(7_200_000L);
- }
-
@DataProvider
public Object[][] validTimeouts() {
return new Object[][]{{"1", 1L}, {" 7200000 ", 7_200_000L},
{String.valueOf(Long.MAX_VALUE), Long.MAX_VALUE}};
@@ -66,6 +54,7 @@ public class ReingestionConsumptionTimeoutTest {
Map<String, String> configs =
Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, timeout);
assertThat(createTimeout(new PinotConfiguration(configs),
Map.of()).getTimeoutMs()).isEqualTo(expectedTimeoutMs);
assertThat(createTimeout(SERVER_CONF,
configs).getTimeoutMs()).isEqualTo(expectedTimeoutMs);
+ assertThat(createTimeout(EMPTY_SERVER_CONF,
configs).getTimeoutMs()).isEqualTo(expectedTimeoutMs);
}
@DataProvider
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceApiTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceApiTest.java
index f4083caf56a..baa6004e733 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceApiTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceApiTest.java
@@ -23,19 +23,27 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nullable;
import javax.ws.rs.client.Entity;
import javax.ws.rs.core.Response;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.core.data.manager.realtime.RealtimeTableDataManager;
import org.apache.pinot.server.api.BaseResourceTest;
+import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
import org.mockito.stubbing.Answer;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static
org.apache.pinot.server.api.resources.ReingestionResource.MAX_PARALLEL_REINGESTIONS;
+import static
org.apache.pinot.server.api.resources.ReingestionResource.waitForCondition;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyString;
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.when;
@@ -47,11 +55,9 @@ public class ReingestionResourceApiTest extends
BaseResourceTest {
// Makes the segment names unique across tests, since a segment is held
until its previous job finishes
private static final AtomicInteger SEQUENCE_NUMBER = new AtomicInteger();
- @Test
- public void testResourceDependenciesAreInjected() {
- // Any request to the resource fails if one of its injected dependencies
is not bound
- Response response =
_webTarget.path("/reingestSegment/jobs").request().get(Response.class);
-
assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+ /// Table availability when a queued re-ingestion starts.
+ private enum TableState {
+ AVAILABLE, REMOVED, SHUT_DOWN
}
@Test
@@ -102,6 +108,78 @@ public class ReingestionResourceApiTest extends
BaseResourceTest {
}
}
+ @DataProvider
+ public Object[][] staleQueuedReingestions() {
+ return new Object[][]{
+ {TableState.AVAILABLE, Status.DONE, null},
+ {TableState.AVAILABLE, null, null},
+ {TableState.AVAILABLE, Status.IN_PROGRESS, null},
+ {TableState.AVAILABLE, Status.COMMITTING, "http://deep-store/segment"},
+ {TableState.REMOVED, Status.COMMITTING, null},
+ {TableState.SHUT_DOWN, Status.COMMITTING, null}
+ };
+ }
+
+ @Test(dataProvider = "staleQueuedReingestions")
+ public void testQueuedReingestionSkipsObsoleteJob(TableState tableState,
@Nullable Status status,
+ @Nullable String downloadUrl)
+ throws Exception {
+ waitForCondition(() -> getJobs().equals("[]"), "previous re-ingestions to
finish", 10, 10_000);
+ CountDownLatch jobsStarted = new CountDownLatch(MAX_PARALLEL_REINGESTIONS);
+ CountDownLatch releaseJobs = new CountDownLatch(1);
+ addTableDataManager(invocation -> {
+ jobsStarted.countDown();
+ releaseJobs.await();
+ throw new IllegalStateException("Finish blocker without building a
segment");
+ });
+
+ String targetRawTableName = "queuedReingestionTable";
+ String targetTableName = targetRawTableName + "_REALTIME";
+ String targetSegmentName = segmentName(targetRawTableName, 0);
+ AtomicReference<SegmentZKMetadata> metadata = new
AtomicReference<>(segmentMetadata(targetSegmentName));
+ RealtimeTableDataManager targetTableDataManager =
mock(RealtimeTableDataManager.class);
+
when(targetTableDataManager.fetchZKMetadata(targetSegmentName)).thenAnswer(invocation
-> metadata.get());
+
when(targetTableDataManager.fetchZKMetadataNullable(targetSegmentName)).thenAnswer(invocation
-> metadata.get());
+ _tableDataManagerMap.put(targetTableName, targetTableDataManager);
+
+ try {
+ for (int partitionId = 0; partitionId < MAX_PARALLEL_REINGESTIONS;
partitionId++) {
+
assertThat(postReingestion(segmentName(partitionId))).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertThat(jobsStarted.await(10, TimeUnit.SECONDS)).isTrue();
+
assertThat(postReingestion(targetSegmentName)).isEqualTo(Response.Status.OK.getStatusCode());
+ assertThat(getJobs()).contains(targetSegmentName);
+
+ // Change the metadata or table availability while the job is queued.
+ SegmentZKMetadata updatedMetadata = null;
+ if (status != null) {
+ updatedMetadata = segmentMetadata(targetSegmentName);
+ updatedMetadata.setStatus(status);
+ updatedMetadata.setDownloadUrl(downloadUrl);
+ }
+ metadata.set(updatedMetadata);
+ if (tableState == TableState.REMOVED) {
+ _tableDataManagerMap.remove(targetTableName);
+ } else if (tableState == TableState.SHUT_DOWN) {
+ when(targetTableDataManager.isShutDown()).thenReturn(true);
+ }
+ releaseJobs.countDown();
+ waitForCondition(() -> !getJobs().contains(targetSegmentName), "queued
re-ingestion to be removed", 10, 10_000);
+
+ verify(targetTableDataManager).fetchZKMetadata(targetSegmentName);
+ verify(targetTableDataManager, tableState == TableState.AVAILABLE ?
times(1) : never())
+ .fetchZKMetadataNullable(targetSegmentName);
+ verify(targetTableDataManager, never()).fetchIndexLoadingConfig();
+ verify(targetTableDataManager, never()).getSegmentBuildSemaphore();
+ verify(_serverInstance.getServerMetrics(), never())
+ .addMeteredTableValue(targetTableName,
ServerMeter.SEGMENT_REINGESTION_FAILURE, 1);
+ } finally {
+ releaseJobs.countDown();
+ _tableDataManagerMap.remove(REINGESTION_REALTIME_TABLE_NAME);
+ _tableDataManagerMap.remove(targetTableName);
+ }
+ }
+
// Runs after the other tests since it stops the admin API
@Test(priority = 1)
public void testRunningReingestionIsStoppedOnShutdown()
@@ -128,21 +206,31 @@ public class ReingestionResourceApiTest extends
BaseResourceTest {
/// fetched on the re-ingestion thread with the given answer.
private void addTableDataManager(Answer<Semaphore>
segmentBuildSemaphoreAnswer) {
RealtimeTableDataManager tableDataManager =
mock(RealtimeTableDataManager.class);
- when(tableDataManager.fetchZKMetadata(anyString())).thenAnswer(invocation
-> {
- SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata((String)
invocation.getArgument(0));
- segmentZKMetadata.setStartOffset("0");
- segmentZKMetadata.setEndOffset("100");
- return segmentZKMetadata;
- });
+ when(tableDataManager.fetchZKMetadata(anyString()))
+ .thenAnswer(invocation -> segmentMetadata(invocation.getArgument(0)));
+ when(tableDataManager.fetchZKMetadataNullable(anyString()))
+ .thenAnswer(invocation -> segmentMetadata(invocation.getArgument(0)));
when(tableDataManager.getSegmentBuildSemaphore()).thenAnswer(segmentBuildSemaphoreAnswer);
_tableDataManagerMap.put(REINGESTION_REALTIME_TABLE_NAME,
tableDataManager);
}
private static String segmentName(int partitionId) {
- return new LLCSegmentName(REINGESTION_RAW_TABLE_NAME, partitionId,
SEQUENCE_NUMBER.getAndIncrement(),
+ return segmentName(REINGESTION_RAW_TABLE_NAME, partitionId);
+ }
+
+ private static String segmentName(String rawTableName, int partitionId) {
+ return new LLCSegmentName(rawTableName, partitionId,
SEQUENCE_NUMBER.getAndIncrement(),
System.currentTimeMillis()).getSegmentName();
}
+ private static SegmentZKMetadata segmentMetadata(String segmentName) {
+ SegmentZKMetadata metadata = new SegmentZKMetadata(segmentName);
+ metadata.setStatus(Status.COMMITTING);
+ metadata.setStartOffset("0");
+ metadata.setEndOffset("100");
+ return metadata;
+ }
+
private int postReingestion(String segmentName) {
try (Response response = _webTarget.path("/reingestSegment/" +
segmentName).request().post(Entity.json(""))) {
return response.getStatus();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
