Jackie-Jiang commented on code in PR #19737:
URL: https://github.com/apache/pinot/pull/19737#discussion_r4253480856
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java:
##########
@@ -202,25 +212,43 @@ public Response reingestSegment(@PathParam("segmentName")
@Encoded String encode
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);
Review Comment:
Non-blocking simplification: the two tracking maps can be consolidated into
one ConcurrentHashMap<String, ReingestionJob> keyed by segment name.
putIfAbsent(segmentName, job) can reject duplicates, values() can serve the
jobs endpoint, and remove(segmentName, job) can handle completion/rejection
cleanup. This removes _reingestingSegments and its per-segment AtomicBoolean;
the job ID can remain in ReingestionJob since nothing looks up the map by job
ID. This duplication is inherited, but the shared job-tracking changes are a
reasonable place to simplify it.
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java:
##########
@@ -202,25 +212,43 @@ public Response reingestSegment(@PathParam("segmentName")
@Encoded String encode
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());
Review Comment:
Follow-up request, not a blocker for this PR: please revalidate segment
metadata when a queued job starts, before constructing the writer, and skip
segments that were deleted or already repaired. The controller can select
different replicas on successive repair passes, so another replica may finish
while this job waits. The captured metadata then causes an unnecessary
consume/build/upload before the controller rejects the already-DONE segment,
occupying a worker slot. Cross-replica duplication and missing revalidation
predate this PR; the shared queue makes the stale-job scenario more visible. A
regression test could queue a job, mark the segment repaired, then release the
worker and verify consumption is skipped.
##########
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);
Review Comment:
Non-blocking test consolidation: testServerConfig repeats the initial
server-only assertion in testClusterConfigChangeIsPickedUpWithoutRestart. Also,
the cluster-over-server assertion in testClusterConfigOverridesServerConfig
overlaps testValidTimeouts. Its distinct cluster-without-server case can move
into testValidTimeouts using EMPTY_SERVER_CONF, allowing the two standalone
methods to be removed while preserving their coverage.
##########
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());
Review Comment:
Non-blocking: this injection smoke test is redundant with the successful
POST/GET requests in the other endpoint tests. Missing resource bindings would
fail those requests too, so this test can be removed without losing coverage.
--
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]
