This is an automated email from the ASF dual-hosted git repository.
9aman 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 ebbc9f46c05 Make pauseless re-ingestion consumption timeout
configurable (#19737)
ebbc9f46c05 is described below
commit ebbc9f46c056e82fd7be74c798e401a17dda5240
Author: Suvodeep Pyne <[email protected]>
AuthorDate: Thu Oct 15 00:39:24 2026 -0700
Make pauseless re-ingestion consumption timeout configurable (#19737)
* Make pauseless re-ingestion consumption timeout configurable
The server's ReingestionResource waited a hard-coded 30 minutes for
re-ingestion consumption to reach the segment end offset. A segment that
cannot be consumed within that window (e.g. ~140M rows at ~49k rows/s)
can never be repaired, because every retry restarts from the start offset.
Add pinot.server.reingestion.consumption.timeoutMs (default 30 minutes,
unchanged). It is resolved when each re-ingestion request is accepted:
the Helix cluster config (picked up live, no restart needed) takes
precedence over the server config, which takes precedence over the
default. Non-numeric or non-positive values are ignored with a warning.
Bind the server's cluster config change handler into the admin API as a
PinotClusterConfigProvider so the resource can read live cluster configs,
and saturate the wait deadline so a large timeout cannot overflow.
* Check re-ingestion completion at the deadline before timing out
waitForCondition slept a fixed check interval (5s) and only re-checked
the condition while the deadline had not passed. With a timeout shorter
than the interval, or completion within the last interval, consumption
that had finished was still reported as a timeout and the segment was
never built or uploaded.
Cap each sleep at the remaining time and check the condition once more
at the deadline before timing out.
* Apply re-ingestion timeout through a cluster config listener and make
ReingestionResource a singleton
Addresses review feedback on the configurable re-ingestion consumption
timeout:
- Resolve the timeout in ReingestionConsumptionTimeout, a
PinotClusterConfigChangeListener registered in
BaseServerStarter and bound into the admin API, instead of reading the
raw cluster config map and the server config
on every request. Values are validated, and invalid ones warned about,
once per config change. The timeout is read
when a job starts consuming.
- Make ReingestionResource a singleton. It was created per request, so
duplicate requests were never rejected, the jobs
listing was always empty, MAX_PARALLEL_REINGESTIONS never applied and
every request leaked a thread pool.
- Report an interrupted wait as an interrupt. The interrupt flag is not
restored, so that closing the segment writer
afterward still waits for its consumer thread before destroying the
segment.
- Drop the always-zero grace period from waitForCondition and take a
BooleanSupplier.
- Document that only the exact key is applied without a restart, the
fallback for a removed or invalid cluster config,
and why the cluster config takes precedence.
- Split the tests into plain unit tests for the timeout and the wait loop,
and endpoint tests for the dependency
injection and the singleton behavior.
* Stop re-ingestion jobs on shutdown, list queued jobs and clarify wait
messages
Addresses the second round of review feedback:
- Shut down the re-ingestion executor when the admin API stops
(@PreDestroy), interrupting running jobs and logging
dropped queued ones. Its threads are now daemon threads, and a submit
rejected during shutdown returns 503 and
clears the in-progress tracking.
- Make StatelessRealtimeSegmentWriter.stopConsumption() wait for the
consumer thread even when the caller is
interrupted, so that close() never destroys the mutable segment while the
consumer thread is still indexing.
- Track a re-ingestion job before submitting it, so that GET
/reingestSegment/jobs also lists jobs waiting for a
re-ingestion thread.
- Name the segment being consumed in the wait messages, and include the
timeout in the timeout message.
- Log the previous and new timeout on a cluster config change, and only
when it changes.
- Widen the timeout of the wait test that checks completion at the deadline
to 4s, so that a pause on a loaded host
cannot make it fail spuriously.
- Move the endpoint tests to the resources package, give each test its own
segment names, and add tests for the
queued job listing and the shutdown.
* Stop re-ingestion consumption at the next batch even if the stream
ignores interrupts
stopConsumption() now waits for the consumer thread uninterruptibly, but
the consumer loop only ended on an
interrupt if the stream plugin threw on it. With a plugin that swallows the
interrupt or retries internally, the
consumer kept consuming to the end offset, so close() and the re-ingestion
shutdown could block until then.
stopConsumption() now also sets a stop flag that the consumer loop checks
before each fetch, so consumption ends at
the next batch (bounded by the fetch timeout) regardless of how the plugin
handles interrupts. A stopped consumption
is not successful and records a cause naming the stop;
getConsumptionException() is marked @Nullable.
Adds the first StatelessRealtimeSegmentWriter unit test: it stops
consumption from a stream that never reaches the end
offset and ignores interrupts, which hung before this change.
---
.../writer/StatelessRealtimeSegmentWriter.java | 20 ++-
.../writer/StatelessRealtimeSegmentWriterTest.java | 168 +++++++++++++++++++++
.../pinot/server/api/AdminApiApplication.java | 4 +-
.../resources/ReingestionConsumptionTimeout.java | 87 +++++++++++
.../server/api/resources/ReingestionResource.java | 112 +++++++++-----
.../server/starter/helix/BaseServerStarter.java | 9 +-
.../apache/pinot/server/api/AccessControlTest.java | 3 +-
.../apache/pinot/server/api/BaseResourceTest.java | 3 +-
.../ReingestionConsumptionTimeoutTest.java | 128 ++++++++++++++++
.../api/resources/ReingestionResourceApiTest.java | 155 +++++++++++++++++++
.../api/resources/ReingestionResourceTest.java | 68 +++++++++
.../apache/pinot/spi/utils/CommonConstants.java | 12 ++
12 files changed, 720 insertions(+), 49 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
index 21961573264..d43975dddac 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java
@@ -19,6 +19,7 @@
package org.apache.pinot.segment.local.realtime.writer;
import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Uninterruptibles;
import java.io.Closeable;
import java.io.File;
import java.time.Duration;
@@ -114,6 +115,8 @@ public class StatelessRealtimeSegmentWriter implements
Closeable {
private final TransformPipeline _transformPipeline;
private volatile boolean _isSuccess = false;
private volatile Throwable _consumptionException;
+ // Set by stopConsumption() to end the consumption at the next batch, even
if the stream ignores the interrupt
+ private volatile boolean _shouldStop;
public StatelessRealtimeSegmentWriter(SegmentZKMetadata segmentZKMetadata,
IndexLoadingConfig indexLoadingConfig,
@Nullable Semaphore segBuildSemaphore)
@@ -259,6 +262,14 @@ public class StatelessRealtimeSegmentWriter implements
Closeable {
_logger.info("Created new consumer thread {} for {}", _consumerThread,
this);
_currentOffset = _startOffset;
while (_currentOffset.compareTo(_endOffset) < 0) {
+ if (_shouldStop) {
+ _consumptionException = new IllegalStateException(
+ "Consumption stopped at offset: " + _currentOffset + " before
reaching end offset: " + _endOffset);
+ _logger.info("Stopped consumption at offset: {} before reaching
end offset: {}", _currentOffset,
+ _endOffset);
+ return;
+ }
+
// Fetch messages
MessageBatch messageBatch = _consumer.fetchMessages(_currentOffset,
_fetchTimeoutMs);
@@ -306,13 +317,11 @@ public class StatelessRealtimeSegmentWriter implements
Closeable {
}
public void stopConsumption() {
+ _shouldStop = true;
if (_consumerThread.isAlive()) {
_consumerThread.interrupt();
- try {
- _consumerThread.join();
- } catch (InterruptedException e) {
- _logger.warn("Interrupted while waiting for consumer thread to
finish");
- }
+ // Wait even if interrupted, so that the segment is not destroyed while
the consumer thread is still indexing
+ Uninterruptibles.joinUninterruptibly(_consumerThread);
}
}
@@ -324,6 +333,7 @@ public class StatelessRealtimeSegmentWriter implements
Closeable {
return _isSuccess;
}
+ @Nullable
public Throwable getConsumptionException() {
return _consumptionException;
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriterTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriterTest.java
new file mode 100644
index 00000000000..9b3101bd860
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriterTest.java
@@ -0,0 +1,168 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.realtime.writer;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.stream.MessageBatch;
+import org.apache.pinot.spi.stream.PartitionGroupConsumer;
+import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
+import org.apache.pinot.spi.stream.StreamConfigProperties;
+import org.apache.pinot.spi.stream.StreamConsumerFactory;
+import org.apache.pinot.spi.stream.StreamMessage;
+import org.apache.pinot.spi.stream.StreamMessageDecoder;
+import org.apache.pinot.spi.stream.StreamMetadataProvider;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/// Tests how [StatelessRealtimeSegmentWriter] stops consuming.
+public class StatelessRealtimeSegmentWriterTest {
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String STREAM_TYPE = "test";
+ // Counted down when the stream is fetched, so that consumption can be
stopped while it is in progress
+ private static final CountDownLatch FETCHED = new CountDownLatch(1);
+
+ @Test(timeOut = 30_000)
+ public void testStopConsumptionWithStreamIgnoringInterrupts()
+ throws Exception {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
+ .addSingleValueDimension("col", DataType.STRING)
+ .build();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME)
+ .setStreamConfigs(Map.of(StreamConfigProperties.STREAM_TYPE,
STREAM_TYPE,
+ StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
StreamConfigProperties.STREAM_TOPIC_NAME),
+ "testTopic",
+ StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
InterruptIgnoringConsumerFactory.class.getName(),
+ StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
StreamConfigProperties.STREAM_DECODER_CLASS),
+ NoOpDecoder.class.getName()))
+ .build();
+ SegmentZKMetadata segmentZKMetadata =
+ new SegmentZKMetadata(new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
System.currentTimeMillis()).getSegmentName());
+ // The stream never reaches the end offset
+ segmentZKMetadata.setStartOffset("0");
+ segmentZKMetadata.setEndOffset("100");
+
+ try (StatelessRealtimeSegmentWriter writer = new
StatelessRealtimeSegmentWriter(segmentZKMetadata,
+ new IndexLoadingConfig(tableConfig, schema), null)) {
+ writer.startConsumption();
+ assertThat(FETCHED.await(10, TimeUnit.SECONDS)).isTrue();
+ writer.stopConsumption();
+ assertThat(writer.isSuccess()).isFalse();
+
assertThat(writer.getConsumptionException()).hasMessageStartingWith("Consumption
stopped at offset");
+ }
+ }
+
+ /// Stream that never returns a message and keeps polling when interrupted.
+ public static class InterruptIgnoringConsumerFactory extends
StreamConsumerFactory {
+ @Nullable
+ @Override
+ public StreamMetadataProvider createPartitionMetadataProvider(String
clientId, int partition) {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public StreamMetadataProvider createStreamMetadataProvider(String
clientId) {
+ return null;
+ }
+
+ @Override
+ public PartitionGroupConsumer createPartitionGroupConsumer(String clientId,
+ PartitionGroupConsumptionStatus partitionGroupConsumptionStatus) {
+ return new PartitionGroupConsumer() {
+ @Override
+ public MessageBatch<byte[]> fetchMessages(StreamPartitionMsgOffset
startOffset, int timeoutMs) {
+ FETCHED.countDown();
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ // Ignore the interrupt like a stream plugin that retries
internally
+ }
+ return new EmptyMessageBatch(startOffset);
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+ }
+ }
+
+ private static class EmptyMessageBatch implements MessageBatch<byte[]> {
+ private final StreamPartitionMsgOffset _nextOffset;
+
+ EmptyMessageBatch(StreamPartitionMsgOffset nextOffset) {
+ _nextOffset = nextOffset;
+ }
+
+ @Override
+ public int getMessageCount() {
+ return 0;
+ }
+
+ @Override
+ public StreamMessage<byte[]> getStreamMessage(int index) {
+ throw new IndexOutOfBoundsException(index);
+ }
+
+ @Override
+ public StreamPartitionMsgOffset getOffsetOfNextBatch() {
+ return _nextOffset;
+ }
+
+ @Override
+ public long getSizeInBytes() {
+ return 0;
+ }
+ }
+
+ public static class NoOpDecoder implements StreamMessageDecoder<byte[]> {
+ @Override
+ public void init(Map<String, String> props, Set<String> fieldsToRead,
String topicName) {
+ }
+
+ @Nullable
+ @Override
+ public GenericRow decode(byte[] payload, GenericRow destination) {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public GenericRow decode(byte[] payload, int offset, int length,
GenericRow destination) {
+ return null;
+ }
+ }
+}
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java
b/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java
index ba33d84aa47..21188b50606 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/api/AdminApiApplication.java
@@ -37,6 +37,7 @@ import org.apache.pinot.core.transport.ListenerConfig;
import org.apache.pinot.core.util.ListenerConfigUtil;
import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache;
import org.apache.pinot.server.access.AccessControlFactory;
+import org.apache.pinot.server.api.resources.ReingestionConsumptionTimeout;
import org.apache.pinot.server.starter.ServerInstance;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
@@ -64,7 +65,7 @@ public class AdminApiApplication extends ResourceConfig {
public AdminApiApplication(ServerInstance instance, AccessControlFactory
accessControlFactory,
- ServerReloadJobStatusCache reloadJobStatusCache,
+ ServerReloadJobStatusCache reloadJobStatusCache,
ReingestionConsumptionTimeout reingestionConsumptionTimeout,
PinotConfiguration serverConf) {
_serverInstance = instance;
_accessControlFactory = accessControlFactory;
@@ -84,6 +85,7 @@ public class AdminApiApplication extends ResourceConfig {
bind(_serverInstance.getServerMetrics()).to(ServerMetrics.class);
bind(_accessControlFactory).to(AccessControlFactory.class);
bind(reloadJobStatusCache).to(ServerReloadJobStatusCache.class);
+
bind(reingestionConsumptionTimeout).to(ReingestionConsumptionTimeout.class);
bind(serverConf.getProperty(CommonConstants.Server.CONFIG_OF_INSTANCE_ID)).named(SERVER_INSTANCE_ID);
String loggerRootDir =
serverConf.getProperty(CommonConstants.Server.CONFIG_OF_LOGGER_ROOT_DIR);
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeout.java
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeout.java
new file mode 100644
index 00000000000..aacfbb36ae4
--- /dev/null
+++
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeout.java
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.server.api.resources;
+
+import com.google.common.primitives.Longs;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.pinot.spi.utils.CommonConstants.Server.CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS;
+import static
org.apache.pinot.spi.utils.CommonConstants.Server.DEFAULT_REINGESTION_CONSUMPTION_TIMEOUT_MS;
+
+
+/// Holds the consumption timeout of the pauseless segment re-ingestion jobs
run by [ReingestionResource]. The cluster
+/// config takes precedence over the server config, which takes precedence
over the default. Cluster config changes are
+/// applied as they arrive. A non-numeric or non-positive value is ignored
with a warning when the config is loaded or
+/// changed.
+///
+/// Thread-safe: cluster config changes are delivered one at a time, and the
latest value is visible to all readers.
+@ThreadSafe
+public class ReingestionConsumptionTimeout implements
PinotClusterConfigChangeListener {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ReingestionConsumptionTimeout.class);
+
+ // Timeout from the server config or the default, used when the cluster
config is absent or invalid
+ private final long _serverTimeoutMs;
+ private volatile long _timeoutMs;
+
+ public ReingestionConsumptionTimeout(PinotConfiguration serverConf) {
+ Long serverTimeoutMs =
+
parseTimeoutMs(serverConf.getProperty(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS),
"server");
+ _serverTimeoutMs = serverTimeoutMs != null ? serverTimeoutMs :
DEFAULT_REINGESTION_CONSUMPTION_TIMEOUT_MS;
+ _timeoutMs = _serverTimeoutMs;
+ LOGGER.info("Initialized re-ingestion consumption timeout to: {}ms",
_timeoutMs);
+ }
+
+ public long getTimeoutMs() {
+ return _timeoutMs;
+ }
+
+ @Override
+ public void onChange(Set<String> changedConfigs, Map<String, String>
clusterConfigs) {
+ if
(!changedConfigs.contains(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS)) {
+ return;
+ }
+ Long clusterTimeoutMs =
parseTimeoutMs(clusterConfigs.get(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS),
"cluster");
+ long oldTimeoutMs = _timeoutMs;
+ _timeoutMs = clusterTimeoutMs != null ? clusterTimeoutMs :
_serverTimeoutMs;
+ if (_timeoutMs != oldTimeoutMs) {
+ LOGGER.info("Updated re-ingestion consumption timeout from: {}ms to:
{}ms", oldTimeoutMs, _timeoutMs);
+ }
+ }
+
+ @Nullable
+ private static Long parseTimeoutMs(@Nullable String value, String
configSource) {
+ if (value == null) {
+ return null;
+ }
+ Long timeoutMs = Longs.tryParse(value.trim());
+ if (timeoutMs == null || timeoutMs <= 0) {
+ LOGGER.warn("Ignoring invalid {} config: {}={}, expecting a positive
number of milliseconds", configSource,
+ CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, value);
+ return null;
+ }
+ return timeoutMs;
+ }
+}
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 f4f08c67054..6264f4c4156 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
@@ -20,7 +20,8 @@ package org.apache.pinot.server.api.resources;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.base.Function;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.math.LongMath;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiKeyAuthDefinition;
@@ -38,10 +39,14 @@ import java.util.UUID;
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;
import javax.ws.rs.Encoded;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
@@ -79,12 +84,13 @@ import static
org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_K
description = "Database context passed through http header. If no
context is provided 'default' database "
+ "context will be considered.")}))
@Path("/")
+@Singleton
public class ReingestionResource {
private static final Logger LOGGER =
LoggerFactory.getLogger(ReingestionResource.class);
// TODO: Make them configurable
- private static final int MAX_PARALLEL_REINGESTIONS =
Math.max(Runtime.getRuntime().availableProcessors() / 2, 1);
- public static final long CONSUMPTION_END_TIMEOUT_MS =
Duration.ofMinutes(30).toMillis();
+ @VisibleForTesting
+ 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
@@ -92,7 +98,7 @@ public class ReingestionResource {
// Executor for asynchronous re-ingestion
private final ExecutorService _reingestionExecutor =
Executors.newFixedThreadPool(MAX_PARALLEL_REINGESTIONS,
- new
ThreadFactoryBuilder().setNameFormat("reingestion-worker-%d").build());
+ new
ThreadFactoryBuilder().setNameFormat("reingestion-worker-%d").setDaemon(true).build());
// Keep track of jobs by jobId => job info
private final ConcurrentHashMap<String, ReingestionJob> _runningJobs = new
ConcurrentHashMap<>();
@@ -100,6 +106,9 @@ public class ReingestionResource {
@Inject
private ServerInstance _serverInstance;
+ @Inject
+ private ReingestionConsumptionTimeout _consumptionTimeout;
+
/// Simple data class to hold job details.
public static class ReingestionJob {
private final String _jobId;
@@ -130,9 +139,10 @@ public class ReingestionResource {
@GET
@Path("/reingestSegment/jobs")
@Produces(MediaType.APPLICATION_JSON)
- @ApiOperation("Get all running re-ingestion jobs along with job IDs")
+ @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
+ // 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();
}
@@ -202,25 +212,43 @@ public class ReingestionResource {
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
- _reingestionExecutor.submit(() -> {
- try {
- _runningJobs.put(jobId, job);
- doReingestSegment(realtimeTableName, segmentZKMetadata,
indexLoadingConfig,
- tableDataManager.getSegmentBuildSemaphore());
- } 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);
- }
- });
+ try {
+ _reingestionExecutor.submit(() -> {
+ try {
+ doReingestSegment(realtimeTableName, segmentZKMetadata,
indexLoadingConfig,
+ tableDataManager.getSegmentBuildSemaphore());
+ } 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);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ // The executor is shut down when the admin API stops
+ _runningJobs.remove(jobId);
+ _reingestingSegments.remove(segmentName);
+ throw new WebApplicationException("Server is shutting down, cannot
re-ingest segment: " + segmentName,
+ Response.Status.SERVICE_UNAVAILABLE);
+ }
return Response.ok(job).build();
}
+ /// Stops the re-ingestion jobs when the admin API stops. Running jobs are
interrupted, and fail and clean up their
+ /// segment. Jobs waiting for a re-ingestion thread are dropped.
+ @PreDestroy
+ public void shutDown() {
+ int numDroppedJobs = _reingestionExecutor.shutdownNow().size();
+ 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,
@@ -229,8 +257,13 @@ public class ReingestionResource {
String segmentName = segmentZKMetadata.getSegmentName();
try (StatelessRealtimeSegmentWriter writer = new
StatelessRealtimeSegmentWriter(segmentZKMetadata,
indexLoadingConfig, segmentBuildSemaphore)) {
+ // 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,
+ consumptionTimeoutMs);
writer.startConsumption();
- waitForCondition((Void) -> writer.isDoneConsuming(), CHECK_INTERVAL_MS,
CONSUMPTION_END_TIMEOUT_MS, 0);
+ waitForCondition(writer::isDoneConsuming, "consumption of segment: " +
segmentName, CHECK_INTERVAL_MS,
+ consumptionTimeoutMs);
writer.stopConsumption();
if (!writer.isSuccess()) {
@@ -247,32 +280,31 @@ public class ReingestionResource {
}
}
- private void waitForCondition(
- Function<Void, Boolean> condition, long checkIntervalMs, long timeoutMs,
long gracePeriodMs) {
- long endTime = System.currentTimeMillis() + timeoutMs;
+ @VisibleForTesting
+ static void waitForCondition(BooleanSupplier condition, String description,
long checkIntervalMs, long timeoutMs) {
+ // Saturate to avoid overflow, since the timeout is configurable
+ long endTime = LongMath.saturatedAdd(System.currentTimeMillis(),
timeoutMs);
- // Adding grace period before starting the condition checks
- if (gracePeriodMs > 0) {
- LOGGER.info("Waiting for a grace period of {} ms before starting
condition checks", gracePeriodMs);
+ while (true) {
try {
- Thread.sleep(gracePeriodMs);
- } catch (InterruptedException e) {
- throw new RuntimeException("Interrupted during grace period wait", e);
- }
- }
-
- while (System.currentTimeMillis() < endTime) {
- try {
- if (Boolean.TRUE.equals(condition.apply(null))) {
- LOGGER.info("Condition satisfied: {}", condition);
+ if (condition.getAsBoolean()) {
+ LOGGER.info("Finished waiting for {}", description);
return;
}
- Thread.sleep(checkIntervalMs);
+ long remainingMs = endTime - System.currentTimeMillis();
+ if (remainingMs <= 0) {
+ break;
+ }
+ // Do not sleep past the deadline, so that the condition is checked
once more at the deadline before timing out
+ Thread.sleep(Math.min(checkIntervalMs, remainingMs));
+ } catch (InterruptedException e) {
+ // Do not restore the interrupt flag, since the interrupt is preserved
as the cause of the converted exception
+ throw new RuntimeException("Interrupted while waiting for " +
description, e);
} catch (Exception e) {
- throw new RuntimeException("Caught exception while checking the
condition", e);
+ throw new RuntimeException("Caught exception while waiting for " +
description, e);
}
}
- throw new RuntimeException("Timeout waiting for condition: " + condition);
+ throw new RuntimeException("Timed out after " + timeoutMs + "ms waiting
for " + description);
}
}
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
index d1abf4f96b2..7d90df3ac06 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
@@ -111,6 +111,7 @@ import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.segment.spi.memory.unsafe.MmapMemoryConfig;
import org.apache.pinot.server.access.AccessControlFactory;
import org.apache.pinot.server.api.AdminApiApplication;
+import org.apache.pinot.server.api.resources.ReingestionConsumptionTimeout;
import org.apache.pinot.server.conf.ServerConf;
import org.apache.pinot.server.realtime.ControllerLeaderLocator;
import org.apache.pinot.server.realtime.ServerSegmentCompletionProtocolHandler;
@@ -186,6 +187,7 @@ public abstract class BaseServerStarter implements
ServiceStartable {
protected volatile boolean _isServerReadyToServeQueries = false;
protected ScheduledExecutorService _helixMessageCountScheduler;
protected ServerReloadJobStatusCache _reloadJobStatusCache;
+ protected ReingestionConsumptionTimeout _reingestionConsumptionTimeout;
// Override this to provide custom thread pool for Helix state transitions.
Null means using Helix's default
@Nullable
protected StateTransitionThreadPoolManager _transitionThreadPoolManager;
@@ -678,6 +680,10 @@ public abstract class BaseServerStarter implements
ServiceStartable {
_clusterConfigChangeHandler.registerClusterConfigChangeListener(_reloadJobStatusCache);
LOGGER.info("Registered ServerReloadJobStatusCache as cluster config
listener");
+ // Register re-ingestion consumption timeout as cluster config listener
for dynamic config updates
+ _reingestionConsumptionTimeout = new
ReingestionConsumptionTimeout(_serverConf);
+
_clusterConfigChangeHandler.registerClusterConfigChangeListener(_reingestionConsumptionTimeout);
+
// install default SSL context if necessary (even if not force-enabled
everywhere)
TlsConfig tlsDefaults = TlsUtils.extractTlsConfig(_serverConf,
Server.SERVER_TLS_PREFIX);
if (StringUtils.isNotBlank(tlsDefaults.getKeyStorePath()) ||
StringUtils.isNotBlank(
@@ -1280,7 +1286,8 @@ public abstract class BaseServerStarter implements
ServiceStartable {
}
protected AdminApiApplication createServerAdminApp() {
- return new AdminApiApplication(_serverInstance, _accessControlFactory,
_reloadJobStatusCache, _serverConf);
+ return new AdminApiApplication(_serverInstance, _accessControlFactory,
_reloadJobStatusCache,
+ _reingestionConsumptionTimeout, _serverConf);
}
/// Creates the [SegmentMessageHandlerFactory] used to handle user-defined
Helix messages for segments.
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java
index c87da3866ef..953d21b6c01 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/AccessControlTest.java
@@ -45,6 +45,7 @@ import org.apache.pinot.server.access.AccessControl;
import org.apache.pinot.server.access.BasicAuthAccessFactory;
import org.apache.pinot.server.access.GrpcRequesterIdentity;
import org.apache.pinot.server.access.HttpRequesterIdentity;
+import org.apache.pinot.server.api.resources.ReingestionConsumptionTimeout;
import org.apache.pinot.server.starter.ServerInstance;
import org.apache.pinot.spi.auth.server.RequesterIdentity;
import org.apache.pinot.spi.env.PinotConfiguration;
@@ -118,7 +119,7 @@ public class AccessControlTest {
ServiceStatus.setServiceStatusCallback(_instanceId, serviceStatusCallback);
_adminApiApplication = new AdminApiApplication(serverInstance,
basicAuthAccessFactory,
- mock(ServerReloadJobStatusCache.class),
+ mock(ServerReloadJobStatusCache.class), new
ReingestionConsumptionTimeout(serverConf),
serverConf);
int adminApiApplicationPort = getAvailablePort();
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java
index 200fcce5cb0..e1387b8fc8a 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/BaseResourceTest.java
@@ -51,6 +51,7 @@ import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver;
import org.apache.pinot.server.access.AllowAllAccessFactory;
+import org.apache.pinot.server.api.resources.ReingestionConsumptionTimeout;
import org.apache.pinot.server.starter.ServerInstance;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -161,7 +162,7 @@ public abstract class BaseResourceTest {
serverConf.setProperty(CommonConstants.Server.CONFIG_OF_INSTANCE_ID,
_instanceId);
configureServerConf(serverConf);
_adminApiApplication = new AdminApiApplication(_serverInstance, new
AllowAllAccessFactory(),
- mock(ServerReloadJobStatusCache.class),
+ mock(ServerReloadJobStatusCache.class), new
ReingestionConsumptionTimeout(serverConf),
serverConf);
_adminApiApplication.start(List.of(
new ListenerConfig(CommonConstants.HTTP_PROTOCOL, "0.0.0.0", 0,
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
new file mode 100644
index 00000000000..2c04697c02f
--- /dev/null
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionConsumptionTimeoutTest.java
@@ -0,0 +1,128 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.server.api.resources;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.helix.model.ClusterConfig;
+import org.apache.pinot.common.config.DefaultClusterConfigChangeHandler;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static
org.apache.pinot.spi.utils.CommonConstants.Server.CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/// Tests that [ReingestionConsumptionTimeout] resolves the timeout from the
cluster config, the server config and the
+/// default, and follows cluster config changes.
+public class ReingestionConsumptionTimeoutTest {
+ private static final PinotConfiguration EMPTY_SERVER_CONF = new
PinotConfiguration();
+ private static final PinotConfiguration SERVER_CONF =
+ new
PinotConfiguration(Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS,
"3600000"));
+ private static final long DEFAULT_TIMEOUT_MS =
Duration.ofMinutes(30).toMillis();
+
+ @Test
+ public void testDefaultsTo30Minutes() {
+ 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}};
+ }
+
+ @Test(dataProvider = "validTimeouts")
+ public void testValidTimeouts(String timeout, long expectedTimeoutMs) {
+ 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);
+ }
+
+ @DataProvider
+ public Object[][] invalidTimeouts() {
+ return new Object[][]{{"abc"}, {""}, {"0"}, {"-1"}, {"1.5"}, {"30m"},
{"9223372036854775808"}};
+ }
+
+ @Test(dataProvider = "invalidTimeouts")
+ public void testInvalidClusterConfigFallsBackToServerConfig(String
invalidTimeout) {
+ Map<String, String> clusterConfigs =
Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, invalidTimeout);
+ assertThat(createTimeout(SERVER_CONF,
clusterConfigs).getTimeoutMs()).isEqualTo(3_600_000L);
+ }
+
+ @Test(dataProvider = "invalidTimeouts")
+ public void testInvalidServerConfigFallsBackToDefault(String invalidTimeout)
{
+ PinotConfiguration serverConf =
+ new
PinotConfiguration(Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS,
invalidTimeout));
+ assertThat(createTimeout(serverConf,
Map.of()).getTimeoutMs()).isEqualTo(DEFAULT_TIMEOUT_MS);
+ }
+
+ @Test
+ public void testClusterConfigChangeIsPickedUpWithoutRestart() {
+ // Registered before the cluster configs are loaded, as on server startup
+ ReingestionConsumptionTimeout timeout = new
ReingestionConsumptionTimeout(SERVER_CONF);
+ DefaultClusterConfigChangeHandler clusterConfigProvider = new
DefaultClusterConfigChangeHandler();
+ clusterConfigProvider.registerClusterConfigChangeListener(timeout);
+ assertThat(timeout.getTimeoutMs()).isEqualTo(3_600_000L);
+
+ setClusterConfigs(clusterConfigProvider,
Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, "10800000"));
+ assertThat(timeout.getTimeoutMs()).isEqualTo(10_800_000L);
+
+ // An invalid cluster config falls back to the server config
+ setClusterConfigs(clusterConfigProvider,
Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, "30m"));
+ assertThat(timeout.getTimeoutMs()).isEqualTo(3_600_000L);
+
+ setClusterConfigs(clusterConfigProvider,
Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, "14400000"));
+ assertThat(timeout.getTimeoutMs()).isEqualTo(14_400_000L);
+
+ // Removing the cluster config falls back to the server config
+ setClusterConfigs(clusterConfigProvider, Map.of());
+ assertThat(timeout.getTimeoutMs()).isEqualTo(3_600_000L);
+ }
+
+ /// Creates the timeout and registers it to a cluster config provider
holding the given cluster configs.
+ private static ReingestionConsumptionTimeout
createTimeout(PinotConfiguration serverConf,
+ Map<String, String> clusterConfigs) {
+ DefaultClusterConfigChangeHandler clusterConfigProvider = new
DefaultClusterConfigChangeHandler();
+ setClusterConfigs(clusterConfigProvider, clusterConfigs);
+ ReingestionConsumptionTimeout timeout = new
ReingestionConsumptionTimeout(serverConf);
+ clusterConfigProvider.registerClusterConfigChangeListener(timeout);
+ return timeout;
+ }
+
+ private static void setClusterConfigs(DefaultClusterConfigChangeHandler
clusterConfigProvider,
+ Map<String, String> clusterConfigs) {
+ ClusterConfig clusterConfig = new ClusterConfig("testCluster");
+ clusterConfig.getRecord().setSimpleFields(new HashMap<>(clusterConfigs));
+ clusterConfigProvider.onClusterConfigChange(clusterConfig, null);
+ }
+}
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
new file mode 100644
index 00000000000..f4083caf56a
--- /dev/null
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceApiTest.java
@@ -0,0 +1,155 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.server.api.resources;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+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.utils.LLCSegmentName;
+import org.apache.pinot.core.data.manager.realtime.RealtimeTableDataManager;
+import org.apache.pinot.server.api.BaseResourceTest;
+import org.mockito.stubbing.Answer;
+import org.testng.annotations.Test;
+
+import static
org.apache.pinot.server.api.resources.ReingestionResource.MAX_PARALLEL_REINGESTIONS;
+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.when;
+
+
+/// Tests that the [ReingestionResource] is served by the admin API with its
injected dependencies, keeps track of the
+/// re-ingestion jobs across requests, and stops them when the admin API stops.
+public class ReingestionResourceApiTest extends BaseResourceTest {
+ private static final String REINGESTION_RAW_TABLE_NAME = "reingestionTable";
+ private static final String REINGESTION_REALTIME_TABLE_NAME =
REINGESTION_RAW_TABLE_NAME + "_REALTIME";
+ // 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());
+ }
+
+ @Test
+ public void testRunningReingestionIsTrackedAcrossRequests()
+ throws Exception {
+ // Hold the job on the re-ingestion thread, where the segment build
semaphore is fetched
+ CountDownLatch jobStarted = new CountDownLatch(1);
+ CountDownLatch releaseJob = new CountDownLatch(1);
+ addTableDataManager(invocation -> {
+ jobStarted.countDown();
+ releaseJob.await();
+ return null;
+ });
+
+ try {
+ String segmentName = segmentName(0);
+
assertThat(postReingestion(segmentName)).isEqualTo(Response.Status.OK.getStatusCode());
+ assertThat(jobStarted.await(10, TimeUnit.SECONDS)).isTrue();
+
+ // Only visible to the next requests when they are served by the same
resource instance
+
assertThat(postReingestion(segmentName)).isEqualTo(Response.Status.CONFLICT.getStatusCode());
+ assertThat(getJobs()).contains(segmentName);
+ } finally {
+ releaseJob.countDown();
+ _tableDataManagerMap.remove(REINGESTION_REALTIME_TABLE_NAME);
+ }
+ }
+
+ @Test
+ public void testQueuedReingestionIsListed() {
+ // Hold every re-ingestion thread so that the last job waits in the queue
+ CountDownLatch releaseJobs = new CountDownLatch(1);
+ addTableDataManager(invocation -> {
+ releaseJobs.await();
+ return null;
+ });
+
+ try {
+ String queuedSegmentName = null;
+ for (int partitionId = 0; partitionId <= MAX_PARALLEL_REINGESTIONS;
partitionId++) {
+ queuedSegmentName = segmentName(partitionId);
+
assertThat(postReingestion(queuedSegmentName)).isEqualTo(Response.Status.OK.getStatusCode());
+ }
+ assertThat(getJobs()).contains(queuedSegmentName);
+ } finally {
+ releaseJobs.countDown();
+ _tableDataManagerMap.remove(REINGESTION_REALTIME_TABLE_NAME);
+ }
+ }
+
+ // Runs after the other tests since it stops the admin API
+ @Test(priority = 1)
+ public void testRunningReingestionIsStoppedOnShutdown()
+ throws Exception {
+ // Hold the job on the re-ingestion thread until it is interrupted
+ AtomicReference<Thread> reingestionThread = new AtomicReference<>();
+ CountDownLatch jobStarted = new CountDownLatch(1);
+ addTableDataManager(invocation -> {
+ reingestionThread.set(Thread.currentThread());
+ jobStarted.countDown();
+ new CountDownLatch(1).await();
+ return null;
+ });
+
assertThat(postReingestion(segmentName(0))).isEqualTo(Response.Status.OK.getStatusCode());
+ assertThat(jobStarted.await(10, TimeUnit.SECONDS)).isTrue();
+
+ _adminApiApplication.stop();
+
+ reingestionThread.get().join(10_000);
+ assertThat(reingestionThread.get().isAlive()).isFalse();
+ }
+
+ /// Adds a table data manager whose segments are pending re-ingestion, and
which answers the segment build semaphore
+ /// 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.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(),
+ System.currentTimeMillis()).getSegmentName();
+ }
+
+ private int postReingestion(String segmentName) {
+ try (Response response = _webTarget.path("/reingestSegment/" +
segmentName).request().post(Entity.json(""))) {
+ return response.getStatus();
+ }
+ }
+
+ private String getJobs() {
+ return
_webTarget.path("/reingestSegment/jobs").request().get(String.class);
+ }
+}
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceTest.java
new file mode 100644
index 00000000000..40d50b692a8
--- /dev/null
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceTest.java
@@ -0,0 +1,68 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.server.api.resources;
+
+import org.testng.annotations.Test;
+
+import static
org.apache.pinot.server.api.resources.ReingestionResource.CHECK_INTERVAL_MS;
+import static
org.apache.pinot.server.api.resources.ReingestionResource.waitForCondition;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/// Tests how [ReingestionResource] waits for the re-ingestion consumption to
complete.
+public class ReingestionResourceTest {
+ private static final String DESCRIPTION = "test condition";
+
+ @Test
+ public void testWaitForConditionWithMaxTimeoutDoesNotTimeOutEarly() {
+ // The condition is false on the first check, so the wait must keep
checking instead of timing out
+ long completionTimeMs = System.currentTimeMillis() + 50;
+ waitForCondition(() -> System.currentTimeMillis() >= completionTimeMs,
DESCRIPTION, 10, Long.MAX_VALUE);
+ }
+
+ @Test
+ public void
testWaitForConditionDetectsCompletionWithTimeoutShorterThanCheckInterval() {
+ // The condition becomes true after the first check. It must be checked
again at the deadline instead of timing out
+ // after sleeping a full check interval.
+ long completionTimeMs = System.currentTimeMillis() + 100;
+ waitForCondition(() -> System.currentTimeMillis() >= completionTimeMs,
DESCRIPTION, CHECK_INTERVAL_MS, 4_000);
+ }
+
+ @Test
+ public void testWaitForConditionDoesNotSleepPastTimeout() {
+ long startTimeMs = System.currentTimeMillis();
+ assertThatThrownBy(() -> waitForCondition(() -> false, DESCRIPTION,
CHECK_INTERVAL_MS, 100)).hasMessage(
+ "Timed out after 100ms waiting for test condition");
+ assertThat(System.currentTimeMillis() -
startTimeMs).isLessThan(CHECK_INTERVAL_MS);
+ }
+
+ @Test
+ public void testWaitForConditionReportsInterrupt() {
+ Thread.currentThread().interrupt();
+ try {
+ assertThatThrownBy(() -> waitForCondition(() -> false, DESCRIPTION,
CHECK_INTERVAL_MS, Long.MAX_VALUE))
+ .hasMessage("Interrupted while waiting for test condition")
+ .hasCauseInstanceOf(InterruptedException.class);
+ } finally {
+ // Clear the interrupt so that it does not leak into other tests
+ Thread.interrupted();
+ }
+ }
+}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index ac0edbd60b3..77a089e0908 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -1822,6 +1822,18 @@ public class CommonConstants {
"ingestion.oom.protection.gcIntervalMs";
public static final long
DEFAULT_SERVER_INGESTION_OOM_PROTECTION_GC_INTERVAL_MS = 30_000L;
+ /// Max time a pauseless segment re-ingestion job (server API `POST
/reingestSegment/{segmentName}`) waits for
+ /// consumption to reach the segment end offset before failing. Read when
each job starts consuming.
+ ///
+ /// Can also be set via cluster config, which is applied without a restart
and takes precedence over the server
+ /// config, so that operators can change the timeout of all servers at
runtime. Notes on the cluster config:
+ /// - Only this exact key is applied without a restart, e.g. not a
`pinot.all.` prefixed or differently cased key.
+ /// - When it is removed or invalid, the server config is used, which
includes the cluster configs copied into it at
+ /// server startup. So to change the timeout, set a new valid value
instead of removing it.
+ public static final String CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS =
+ "pinot.server.reingestion.consumption.timeoutMs";
+ public static final long DEFAULT_REINGESTION_CONSUMPTION_TIMEOUT_MS =
1_800_000L; // 30 minutes
+
public static final String CONFIG_OF_MMAP_DEFAULT_ADVICE =
"pinot.server.mmap.advice.default";
public static final String PREFIX_OF_CONFIG_OF_SEGMENT_FETCHER_FACTORY =
"pinot.server.segment.fetcher";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
