This is an automated email from the ASF dual-hosted git repository.
mcvsubbu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new c238c2b Added tests for RealtimeSegmentValidationManager (#4306)
c238c2b is described below
commit c238c2b66a4fcc30c0894a465bce4ae7c1a2b800
Author: Subbu Subramaniam <[email protected]>
AuthorDate: Wed Jun 12 15:52:58 2019 -0700
Added tests for RealtimeSegmentValidationManager (#4306)
* Added tests for RealtimeSegmentValidationManager
Removed the dropping of realtime table and setting it up again. Relying on
the realtime table being created ahead of time
* Remove configurations in favor of overridden methods for test.
We seem to be introducing extra configurations just for testing purpose.
Removed the extra configs for initial delay in favor of overridden methods
for tests.
Retained the configuration for frequency, since an admin may want to
set different frequencies for the different tasks depending on the
cluster. In real clusters, the initial delay is a random one subject
to a minimum
---
.../apache/pinot/controller/ControllerConf.java | 19 ++----
.../helix/core/minion/PinotTaskManager.java | 2 +-
.../RealtimeSegmentValidationManager.java | 7 ++-
.../ControllerPeriodicTasksIntegrationTests.java | 71 +++++++++++++++++-----
4 files changed, 67 insertions(+), 32 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
index d5a620e..697fcf3 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
@@ -463,11 +463,6 @@ public class ControllerConf extends
PropertiesConfiguration {
getPeriodicTaskInitialDelayInSeconds());
}
- public void setBrokerResourceValidationInitialDelayInSeconds(long
validationInitialDelayInSeconds) {
-
setProperty(ControllerPeriodicTasksConf.BROKER_RESOURCE_VALIDATION_INITIAL_DELAY_IN_SECONDS,
- validationInitialDelayInSeconds);
- }
-
public int getStatusCheckerFrequencyInSeconds() {
if
(containsKey(ControllerPeriodicTasksConf.STATUS_CHECKER_FREQUENCY_IN_SECONDS)) {
return Integer.parseInt((String)
getProperty(ControllerPeriodicTasksConf.STATUS_CHECKER_FREQUENCY_IN_SECONDS));
@@ -620,18 +615,12 @@ public class ControllerConf extends
PropertiesConfiguration {
ControllerPeriodicTasksConf.getRandomInitialDelayInSeconds());
}
- public void setStatusCheckerInitialDelayInSeconds(long
initialDelayInSeconds) {
-
setProperty(ControllerPeriodicTasksConf.STATUS_CHECKER_INITIAL_DELAY_IN_SECONDS,
initialDelayInSeconds);
- }
-
- public void setRealtimeSegmentRelocationInitialDelayInSeconds(long
initialDelayInSeconds) {
-
setProperty(ControllerPeriodicTasksConf.REALTIME_SEGMENT_RELOCATION_INITIAL_DELAY_IN_SECONDS,
- initialDelayInSeconds);
+ public long getRealtimeSegmentValidationManagerInitialDelaySeconds() {
+ return getPeriodicTaskInitialDelayInSeconds();
}
- public void setOfflineSegmentIntervalCheckerInitialDelayInSeconds(long
initialDelayInSeconds) {
-
setProperty(ControllerPeriodicTasksConf.OFFLINE_SEGMENT_INTERVAL_CHECKER_INITIAL_DELAY_IN_SECONDS,
- initialDelayInSeconds);
+ public long getPinotTaskManagerInitialDelaySeconds() {
+ return getPeriodicTaskInitialDelayInSeconds();
}
public long getPeriodicTaskInitialDelayInSeconds() {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
index 6bb5c9d..7a09c00 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
@@ -54,7 +54,7 @@ public class PinotTaskManager extends
ControllerPeriodicTask<Void> {
PinotHelixResourceManager helixResourceManager, ControllerConf
controllerConf,
ControllerMetrics controllerMetrics) {
super("PinotTaskManager",
controllerConf.getTaskManagerFrequencyInSeconds(),
- controllerConf.getPeriodicTaskInitialDelayInSeconds(),
helixResourceManager, controllerMetrics);
+ controllerConf.getPinotTaskManagerInitialDelaySeconds(),
helixResourceManager, controllerMetrics);
_helixTaskResourceManager = helixTaskResourceManager;
_clusterInfoProvider = new ClusterInfoProvider(helixResourceManager,
helixTaskResourceManager, controllerConf);
_taskGeneratorRegistry = new TaskGeneratorRegistry(_clusterInfoProvider);
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
index 6c0b59f..5eb5a6f 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
@@ -56,7 +56,7 @@ public class RealtimeSegmentValidationManager extends
ControllerPeriodicTask<Rea
PinotLLCRealtimeSegmentManager llcRealtimeSegmentManager,
ValidationMetrics validationMetrics,
ControllerMetrics controllerMetrics) {
super("RealtimeSegmentValidationManager",
config.getRealtimeSegmentValidationFrequencyInSeconds(),
- config.getPeriodicTaskInitialDelayInSeconds(),
pinotHelixResourceManager, controllerMetrics);
+ config.getRealtimeSegmentValidationManagerInitialDelaySeconds(),
pinotHelixResourceManager, controllerMetrics);
_llcRealtimeSegmentManager = llcRealtimeSegmentManager;
_validationMetrics = validationMetrics;
@@ -156,4 +156,9 @@ public class RealtimeSegmentValidationManager extends
ControllerPeriodicTask<Rea
public static final class Context {
private boolean _updateRealtimeDocumentCount;
}
+
+ @VisibleForTesting
+ public ValidationMetrics getValidationMetrics() {
+ return _validationMetrics;
+ }
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ControllerPeriodicTasksIntegrationTests.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ControllerPeriodicTasksIntegrationTests.java
index a0a2c5c..ccd76b4 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ControllerPeriodicTasksIntegrationTests.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ControllerPeriodicTasksIntegrationTests.java
@@ -49,6 +49,7 @@ import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.common.utils.retry.RetryPolicies;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.validation.OfflineSegmentIntervalChecker;
+import org.apache.pinot.controller.validation.RealtimeSegmentValidationManager;
import org.apache.pinot.core.indexsegment.generator.SegmentVersion;
import org.apache.pinot.core.realtime.impl.kafka.KafkaStarterUtils;
import org.apache.pinot.util.TestUtils;
@@ -79,6 +80,30 @@ import org.testng.annotations.Test;
*/
public class ControllerPeriodicTasksIntegrationTests extends
BaseClusterIntegrationTestSet {
+ // Controller configuration used in this class.
+ private class TestControllerConf extends ControllerConf {
+ private TestControllerConf(ControllerConf controllerConf) {
+ copy(controllerConf);
+ }
+
+ @Override
+ public long getRealtimeSegmentValidationManagerInitialDelaySeconds() {
+ return PERIODIC_TASK_INITIAL_DELAY_SECONDS;
+ }
+ public long getStatusCheckerInitialDelayInSeconds() {
+ return PERIODIC_TASK_INITIAL_DELAY_SECONDS;
+ }
+ public long getRealtimeSegmentRelocationInitialDelayInSeconds() {
+ return PERIODIC_TASK_INITIAL_DELAY_SECONDS;
+ }
+ public long getBrokerResourceValidationInitialDelayInSeconds() {
+ return PERIODIC_TASK_INITIAL_DELAY_SECONDS;
+ }
+ public long getOfflineSegmentIntervalCheckerInitialDelayInSeconds() {
+ return PERIODIC_TASK_INITIAL_DELAY_SECONDS;
+ }
+ }
+
private static final String TENANT_NAME = "TestTenant";
private static final String DEFAULT_TABLE_NAME = "mytable";
@@ -102,16 +127,13 @@ public class ControllerPeriodicTasksIntegrationTests
extends BaseClusterIntegrat
// Set initial delay of 60 seconds for periodic tasks, to allow time for
tables setup.
// Run at 5 seconds freq in order to keep them running, in case first run
happens before table setup
- ControllerConf controllerConf = getDefaultControllerConfiguration();
+ TestControllerConf controllerConf = new
TestControllerConf(getDefaultControllerConfiguration());
controllerConf.setTenantIsolationEnabled(false);
-
controllerConf.setStatusCheckerInitialDelayInSeconds(PERIODIC_TASK_INITIAL_DELAY_SECONDS);
controllerConf.setStatusCheckerFrequencyInSeconds(PERIODIC_TASK_FREQ_SECONDS);
-
controllerConf.setRealtimeSegmentRelocationInitialDelayInSeconds(PERIODIC_TASK_INITIAL_DELAY_SECONDS);
controllerConf.setRealtimeSegmentRelocatorFrequency(PERIODIC_TASK_FREQ);
-
controllerConf.setBrokerResourceValidationInitialDelayInSeconds(PERIODIC_TASK_INITIAL_DELAY_SECONDS);
controllerConf.setBrokerResourceValidationFrequencyInSeconds(PERIODIC_TASK_FREQ_SECONDS);
-
controllerConf.setOfflineSegmentIntervalCheckerInitialDelayInSeconds(PERIODIC_TASK_INITIAL_DELAY_SECONDS);
controllerConf.setOfflineSegmentIntervalCheckerFrequencyInSeconds(PERIODIC_TASK_FREQ_SECONDS);
+
controllerConf.setRealtimeSegmentValidationFrequencyInSeconds(PERIODIC_TASK_FREQ_SECONDS);
startController(controllerConf);
startBroker();
@@ -331,7 +353,6 @@ public class ControllerPeriodicTasksIntegrationTests
extends BaseClusterIntegrat
dropOfflineTable(emptyTable);
dropOfflineTable(disabledOfflineTable);
dropOfflineTable(errorOfflineTable);
- dropOfflineTable(basicRealtimeTable);
}
/**
@@ -344,9 +365,6 @@ public class ControllerPeriodicTasksIntegrationTests
extends BaseClusterIntegrat
String relocationTable = getDefaultRealtimeTableName();
context.setAttribute("relocationTable", relocationTable);
- // setup default realtime table
- setupRealtimeTable(relocationTable, getKafkaTopic(), _avroFiles.get(0));
-
// add tag override for relocation
TenantConfig tenantConfig = new TenantConfig();
tenantConfig.setServer(TENANT_NAME);
@@ -391,12 +409,6 @@ public class ControllerPeriodicTasksIntegrationTests
extends BaseClusterIntegrat
Assert.assertTrue(Collections.disjoint(consuming, completed));
}
- @AfterGroups(groups = "realtimeSegmentRelocator", dependsOnGroups =
"segmentStatusChecker")
- public void afterRealtimeSegmentRelocatorTest(ITestContext context) throws
Exception {
- String relocationTable = (String) context.getAttribute("relocationTable");
- dropRealtimeTable(relocationTable);
- }
-
@BeforeGroups(groups = "brokerResourceValidationManager", dependsOnGroups =
"realtimeSegmentRelocator")
public void beforeBrokerResourceValidationManagerTest(ITestContext context)
throws Exception {
@@ -499,6 +511,35 @@ public class ControllerPeriodicTasksIntegrationTests
extends BaseClusterIntegrat
115545);
}
+ /**
+ * Group - realtimeSegmentValidationManager - Integration tests for {@link
org.apache.pinot.controller.validation.RealtimeSegmentValidationManager}
+ * @param context
+ * @throws Exception
+ */
+
+ @Test(groups="realtimeSegmentValidationManager", dependsOnGroups =
"offlineSegmentIntervalChecker")
+ public void testRealtimeSegmentValidationManager(ITestContext context)
throws Exception {
+ ControllerMetrics controllerMetrics =
_controllerStarter.getControllerMetrics();
+ long taskRunCount =
controllerMetrics.getMeteredTableValue("RealtimeSegmentValidationManager",
+ ControllerMeter.CONTROLLER_PERIODIC_TASK_RUN).count();
+
+ // Wait until the RealtimeSegmentValidationManager runs at least once.
Most likely it already ran once
+ // on the realtime table (default one) already setup, so we should have
the total document count on that
+ // realtime table.
+ TestUtils.waitForCondition(input ->
+
controllerMetrics.getMeteredTableValue("RealtimeSegmentValidationManager",
ControllerMeter.CONTROLLER_PERIODIC_TASK_RUN)
+ .count() > taskRunCount, 60_000, "Timed out waiting for
RealtimeSegmentValidationManager to run");
+
+
Assert.assertTrue(controllerMetrics.getValueOfGlobalGauge(ControllerGauge.PERIODIC_TASK_NUM_TABLES_PROCESSED,
+ "RealtimeSegmentValidationManager") > 0);
+ RealtimeSegmentValidationManager validationManager =
_controllerStarter.getRealtimeSegmentValidationManager();
+ ValidationMetrics validationMetrics =
validationManager.getValidationMetrics();
+ // Make sure we processed the realtime table to get the total document
count. Should have been done the first
+ // time RealtimeSegmentValidationManager ran on the default realtime table.
+
Assert.assertTrue(validationMetrics.getValueOfGauge(ValidationMetrics.makeGaugeName(getDefaultRealtimeTableName(),
+ "TotalDocumentCount")) > 0);
+ }
+
// TODO: tests for other ControllerPeriodicTasks (RetentionManagert ,
RealtimeSegmentValidationManager)
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
