suvodeep-pyne commented on code in PR #19737:
URL: https://github.com/apache/pinot/pull/19737#discussion_r4201360966
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java:
##########
@@ -189,6 +204,8 @@ public Response reingestSegment(@PathParam("segmentName")
@Encoded String encode
}
IndexLoadingConfig indexLoadingConfig =
tableDataManager.fetchIndexLoadingConfig();
+ long consumptionTimeoutMs = getConsumptionTimeoutMs(_clusterConfigProvider,
+ (PinotConfiguration)
_application.getProperties().get(AdminApiApplication.PINOT_CONFIGURATION));
// Check if this segment is already being re-ingested
AtomicBoolean isIngesting =
_reingestingSegments.computeIfAbsent(segmentName, k -> new
AtomicBoolean(false));
Review Comment:
Fixed in 7c753f3b71. `ReingestionResource` is now `@Singleton`, like
`ControllerJobStatusResource`, so the in-progress map, the jobs listing and the
executor are shared across requests. A duplicate POST for a segment that is
queued or running on the same server now gets 409; `repairSegmentsInErrorState`
catches that per segment, logs it and moves on to the next one. One visible
side effect: concurrent re-ingestions per server are now actually capped at
`MAX_PARALLEL_REINGESTIONS`, and extra jobs queue. The timeout is read when a
job starts consuming, so time spent queued doesn't count against it.
`ReingestionResourceApiTest.testRunningReingestionIsTrackedAcrossRequests`
holds a job running and checks the 409 and the jobs listing. It fails without
`@Singleton`.
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java:
##########
@@ -261,13 +309,18 @@ private void waitForCondition(
}
}
- while (System.currentTimeMillis() < endTime) {
+ while (true) {
try {
if (Boolean.TRUE.equals(condition.apply(null))) {
LOGGER.info("Condition satisfied: {}", condition);
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));
Review Comment:
Partly done in 7c753f3b71. `InterruptedException` now has its own catch, and
the message says the wait was interrupted, so it's no longer reported as a
failed condition check. I deliberately didn't restore the flag, though.
`waitForCondition` throws inside the try-with-resources in `doReingestSegment`,
so `writer.close()` runs next and calls `stopConsumption()`, which interrupts
the consumer thread and then `join()`s it. With the caller's flag set, that
`join()` throws right away and only logs a warning. `close()` would then
destroy the mutable segment and delete its dirs while the consumer thread may
still be indexing into it. The interrupt is still kept as the cause, the task
ends right after, and nothing on that path reads the flag. This also follows
C5.26 in `kb/code-review-principles.md`. The grace-period block is gone (see
the `gracePeriodMs` thread). `testWaitForConditionReportsInterrupt` pins both
the message and the cleared flag.
##########
pinot-server/src/test/java/org/apache/pinot/server/api/resources/ReingestionResourceTest.java:
##########
@@ -0,0 +1,156 @@
+/**
+ * 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.HashMap;
+import java.util.Map;
+import javax.ws.rs.core.Response;
+import org.apache.helix.model.ClusterConfig;
+import org.apache.pinot.common.config.DefaultClusterConfigChangeHandler;
+import org.apache.pinot.server.api.BaseResourceTest;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.testng.annotations.DataProvider;
+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.getConsumptionTimeoutMs;
+import static
org.apache.pinot.server.api.resources.ReingestionResource.waitForCondition;
+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;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/// Tests the consumption timeout resolution of [ReingestionResource] and the
wiring of its injected dependencies.
+public class ReingestionResourceTest extends BaseResourceTest {
+ 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"));
+
+ @Test
+ public void testResourceDependenciesAreInjected() {
+ // The resource is created per request, so any request 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 testConsumptionTimeoutDefaultsTo30Minutes() {
+ assertThat(DEFAULT_REINGESTION_CONSUMPTION_TIMEOUT_MS).isEqualTo(30 * 60 *
1000L);
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider(Map.of()),
EMPTY_SERVER_CONF)).isEqualTo(
+ DEFAULT_REINGESTION_CONSUMPTION_TIMEOUT_MS);
+ }
+
+ @Test
+ public void testConsumptionTimeoutFromServerConfig() {
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider(Map.of()),
SERVER_CONF)).isEqualTo(3_600_000L);
+ }
+
+ @Test
+ public void testClusterConfigOverridesServerConfig() {
+ DefaultClusterConfigChangeHandler clusterConfigProvider =
+
clusterConfigProvider(Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS,
"7200000"));
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider,
SERVER_CONF)).isEqualTo(7_200_000L);
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider,
EMPTY_SERVER_CONF)).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) {
+ PinotConfiguration serverConf =
+ new
PinotConfiguration(Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS,
timeout));
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider(Map.of()),
serverConf)).isEqualTo(expectedTimeoutMs);
+ assertThat(getConsumptionTimeoutMs(
+
clusterConfigProvider(Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS,
timeout)), SERVER_CONF)).isEqualTo(
+ expectedTimeoutMs);
+ }
+
+ @DataProvider
+ public Object[][] invalidTimeouts() {
+ return new Object[][]{{"abc"}, {""}, {"0"}, {"-1"}, {"1.5"}, {"30m"},
{"9223372036854775808"}};
+ }
+
+ @Test(dataProvider = "invalidTimeouts")
+ public void testInvalidClusterConfigFallsBackToServerConfig(String
invalidTimeout) {
+ DefaultClusterConfigChangeHandler clusterConfigProvider =
+
clusterConfigProvider(Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS,
invalidTimeout));
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider,
SERVER_CONF)).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(getConsumptionTimeoutMs(clusterConfigProvider(Map.of()),
serverConf)).isEqualTo(
+ DEFAULT_REINGESTION_CONSUMPTION_TIMEOUT_MS);
+ }
+
+ @Test
+ public void testClusterConfigChangeIsPickedUpWithoutRestart() {
+ DefaultClusterConfigChangeHandler clusterConfigProvider =
clusterConfigProvider(Map.of());
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider,
SERVER_CONF)).isEqualTo(3_600_000L);
+
+ // Set the cluster config on the same provider
+ setClusterConfigs(clusterConfigProvider,
Map.of(CONFIG_OF_REINGESTION_CONSUMPTION_TIMEOUT_MS, "10800000"));
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider,
SERVER_CONF)).isEqualTo(10_800_000L);
+
+ // Remove the cluster config, which falls back to the server config
+ setClusterConfigs(clusterConfigProvider, Map.of());
+ assertThat(getConsumptionTimeoutMs(clusterConfigProvider,
SERVER_CONF)).isEqualTo(3_600_000L);
+ }
+
+ @Test
+ public void testWaitForConditionWithMaxTimeoutDoesNotOverflow() {
+ // Without saturation, the deadline overflows to the past and the wait
times out before checking the condition
+ waitForCondition(v -> true, 1, Long.MAX_VALUE, 0);
Review Comment:
Rewritten as suggested in 7c753f3b71: the condition is false on the first
check and true after ~50 ms, with a `Long.MAX_VALUE` timeout. One correction
from checking it, though: that test also passes with a plain `+`. The loop
computes `endTime - now`, which stays exact under two's-complement wraparound
(the same idiom as `System.nanoTime()`). An overflowed deadline only breaks a
`now >= endTime` style check like the original loop's, and I confirmed the test
fails against that. I kept `saturatedAdd` so `endTime` stays a real timestamp,
and reworded the test comment so it doesn't claim more than the test checks.
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ReingestionResource.java:
##########
@@ -189,6 +204,8 @@ public Response reingestSegment(@PathParam("segmentName")
@Encoded String encode
}
IndexLoadingConfig indexLoadingConfig =
tableDataManager.fetchIndexLoadingConfig();
+ long consumptionTimeoutMs = getConsumptionTimeoutMs(_clusterConfigProvider,
+ (PinotConfiguration)
_application.getProperties().get(AdminApiApplication.PINOT_CONFIGURATION));
Review Comment:
Done in 7c753f3b71, together with the listener suggestion. The resource no
longer reads `PINOT_CONFIGURATION` or casts anything. It `@Inject`s a typed
`ReingestionConsumptionTimeout`, bound in `AdminApiApplication`, which reads
the server config once when `BaseServerStarter` builds it.
`ReingestionResourceApiTest.testResourceDependenciesAreInjected` fails if the
binding is missing (checked by removing it).
--
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]
