You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2019/07/29 02:03:45 UTC
[incubator-pinot] branch clean_up_test_util created (now 5cb9d92)
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a change to branch clean_up_test_util
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
at 5cb9d92 Clean up test utils, disconnect fake instances
This branch includes the following new commits:
new 5cb9d92 Clean up test utils, disconnect fake instances
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/01: Clean up test utils,
disconnect fake instances
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch clean_up_test_util
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 5cb9d92fb4ea833ba1dfe4977980036278254225
Author: Jackie (Xiaotian) Jiang <xa...@linkedin.com>
AuthorDate: Sun Jul 28 19:00:36 2019 -0700
Clean up test utils, disconnect fake instances
- Remove redundant util classes
- Move controller test related methods into ControllerTest for easier usage
- Disconnect fake instances created in the test
- No change for the test logic
---
.../broker/broker/HelixBrokerStarterTest.java | 25 +-
.../helix/ControllerRequestBuilderUtil.java | 152 -----------
.../EmptyBrokerOnlineOfflineStateModelFactory.java | 70 -----
...EmptySegmentOnlineOfflineStateModelFactory.java | 81 ------
.../pinot/controller/api/PinotFileUploadTest.java | 39 ++-
.../api/PinotSegmentRestletResourceTest.java | 35 +--
.../api/PinotTableRestletResourceTest.java | 11 +-
.../api/PinotTenantRestletResourceTest.java | 12 +-
.../pinot/controller/api/TableViewsTest.java | 12 +-
.../helix/ControllerInstanceToggleTest.java | 12 +-
.../controller/helix/ControllerSentinelTestV2.java | 23 +-
.../controller/helix/ControllerTenantTest.java | 27 +-
.../pinot/controller/helix/ControllerTest.java | 286 +++++++++++++++++++--
.../pinot/controller/helix/HelixHelperTest.java | 6 +-
.../controller/helix/PinotControllerModeTest.java | 22 +-
.../controller/helix/PinotResourceManagerTest.java | 48 ++--
.../helix/core/PinotHelixResourceManagerTest.java | 26 +-
.../ReplicaGroupRebalanceStrategyTest.java | 77 +++---
.../sharding/SegmentAssignmentStrategyTest.java | 145 ++++-------
.../validation/ValidationManagerTest.java | 47 ++--
...umSegmentAssignmentStrategyIntegrationTest.java | 8 +-
.../tests/BaseClusterIntegrationTest.java | 20 +-
.../tests/BaseClusterIntegrationTestSet.java | 17 +-
.../pinot/integration/tests/ClusterTest.java | 22 +-
.../ControllerPeriodicTasksIntegrationTests.java | 97 ++++---
.../tests/OfflineClusterIntegrationTest.java | 14 +-
.../tests/PinotURIUploadIntegrationTest.java | 4 +-
.../tests/SegmentCompletionIntegrationTest.java | 10 +-
.../perf/BenchmarkRealtimeConsumptionSpeed.java | 4 +-
.../org/apache/pinot/perf/RealtimeStressTest.java | 4 +-
.../server/starter/helix/HelixServerStarter.java | 6 +-
31 files changed, 587 insertions(+), 775 deletions(-)
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java
index 8ecff02..d903626 100644
--- a/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java
@@ -41,7 +41,6 @@ import org.apache.pinot.common.utils.CommonConstants.Broker;
import org.apache.pinot.common.utils.CommonConstants.Helix;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
import org.apache.pinot.common.utils.ZkStarter;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.apache.pinot.util.TestUtils;
@@ -74,14 +73,11 @@ public class HelixBrokerStarterTest extends ControllerTest {
Configuration brokerConf = new BaseConfiguration();
brokerConf.addProperty(Helix.KEY_OF_BROKER_QUERY_PORT, 18099);
brokerConf.addProperty(Broker.CONFIG_OF_BROKER_REFRESH_TIMEBOUNDARY_INFO_SLEEP_INTERVAL, 100L);
- _brokerStarter = new HelixBrokerStarter(brokerConf, getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR);
+ _brokerStarter = new HelixBrokerStarter(brokerConf, _helixClusterName, ZkStarter.DEFAULT_ZK_STR);
_brokerStarter.start();
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR, NUM_BROKERS - 1,
- true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR, NUM_SERVERS, true);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_BROKERS - 1, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_SERVERS, true);
Schema schema = new Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
.addTime(TIME_COLUMN_NAME, TimeUnit.DAYS, FieldSpec.DataType.INT).build();
@@ -104,7 +100,7 @@ public class HelixBrokerStarterTest extends ControllerTest {
TestUtils.waitForCondition(aVoid -> {
ExternalView offlineTableExternalView =
- _helixAdmin.getResourceExternalView(getHelixClusterName(), OFFLINE_TABLE_NAME);
+ _helixAdmin.getResourceExternalView(_helixClusterName, OFFLINE_TABLE_NAME);
return offlineTableExternalView != null
&& offlineTableExternalView.getPartitionSet().size() == NUM_OFFLINE_SEGMENTS;
}, 30_000L, "Failed to find all OFFLINE segments in the ExternalView");
@@ -124,16 +120,16 @@ public class HelixBrokerStarterTest extends ControllerTest {
public void testResourceAndTagAssignment()
throws Exception {
assertEquals(
- _helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), TagNameUtils.getBrokerTagForTenant(null))
- .size(), NUM_BROKERS);
+ _helixAdmin.getInstancesInClusterWithTag(_helixClusterName, TagNameUtils.getBrokerTagForTenant(null)).size(),
+ NUM_BROKERS);
IdealState brokerResourceIdealState =
- _helixAdmin.getResourceIdealState(getHelixClusterName(), Helix.BROKER_RESOURCE_INSTANCE);
+ _helixAdmin.getResourceIdealState(_helixClusterName, Helix.BROKER_RESOURCE_INSTANCE);
assertEquals(brokerResourceIdealState.getInstanceSet(OFFLINE_TABLE_NAME).size(), NUM_BROKERS);
assertEquals(brokerResourceIdealState.getInstanceSet(REALTIME_TABLE_NAME).size(), NUM_BROKERS);
ExternalView brokerResourceExternalView =
- _helixAdmin.getResourceExternalView(getHelixClusterName(), Helix.BROKER_RESOURCE_INSTANCE);
+ _helixAdmin.getResourceExternalView(_helixClusterName, Helix.BROKER_RESOURCE_INSTANCE);
assertEquals(brokerResourceExternalView.getStateMap(OFFLINE_TABLE_NAME).size(), NUM_BROKERS);
assertEquals(brokerResourceExternalView.getStateMap(REALTIME_TABLE_NAME).size(), NUM_BROKERS);
@@ -166,12 +162,12 @@ public class HelixBrokerStarterTest extends ControllerTest {
assertNotNull(newTableConfigInCluster);
assertEquals(newTableConfigInCluster.getTenantConfig().getBroker(), TagNameUtils.DEFAULT_TENANT_NAME);
- brokerResourceIdealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), Helix.BROKER_RESOURCE_INSTANCE);
+ brokerResourceIdealState = _helixAdmin.getResourceIdealState(_helixClusterName, Helix.BROKER_RESOURCE_INSTANCE);
assertEquals(brokerResourceIdealState.getInstanceSet(newOfflineTableName).size(), NUM_BROKERS);
TestUtils.waitForCondition(aVoid -> {
Map<String, String> newTableStateMap =
- _helixAdmin.getResourceExternalView(getHelixClusterName(), Helix.BROKER_RESOURCE_INSTANCE)
+ _helixAdmin.getResourceExternalView(_helixClusterName, Helix.BROKER_RESOURCE_INSTANCE)
.getStateMap(newOfflineTableName);
return newTableStateMap != null && newTableStateMap.size() == NUM_BROKERS;
}, 30_000L, "Failed to find all brokers for the new table in the brokerResource ExternalView");
@@ -207,6 +203,7 @@ public class HelixBrokerStarterTest extends ControllerTest {
@AfterClass
public void tearDown() {
+ stopFakeInstances();
_brokerStarter.shutdown();
stopController();
stopZk();
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestBuilderUtil.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestBuilderUtil.java
deleted file mode 100644
index bedc11a..0000000
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/ControllerRequestBuilderUtil.java
+++ /dev/null
@@ -1,152 +0,0 @@
-/**
- * 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.controller.helix;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.helix.HelixManager;
-import org.apache.helix.HelixManagerFactory;
-import org.apache.helix.InstanceType;
-import org.apache.helix.model.HelixConfigScope;
-import org.apache.helix.model.builder.HelixConfigScopeBuilder;
-import org.apache.helix.participant.StateMachineEngine;
-import org.apache.helix.participant.statemachine.StateModelFactory;
-import org.apache.pinot.common.config.TableNameBuilder;
-import org.apache.pinot.common.config.TagNameUtils;
-import org.apache.pinot.common.config.Tenant;
-import org.apache.pinot.common.config.Tenant.TenantBuilder;
-import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.JsonUtils;
-import org.apache.pinot.common.utils.TenantRole;
-
-import static org.apache.pinot.common.utils.CommonConstants.Helix.UNTAGGED_BROKER_INSTANCE;
-import static org.apache.pinot.common.utils.CommonConstants.Helix.UNTAGGED_SERVER_INSTANCE;
-
-
-public class ControllerRequestBuilderUtil {
- private ControllerRequestBuilderUtil() {
- }
-
- public static void addFakeBrokerInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- int numInstances)
- throws Exception {
- addFakeBrokerInstancesToAutoJoinHelixCluster(helixClusterName, zkServer, numInstances, false);
- }
-
- public static void addFakeBrokerInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- int numInstances, boolean isSingleTenant)
- throws Exception {
- for (int i = 0; i < numInstances; ++i) {
- final String brokerId = "Broker_localhost_" + i;
- final HelixManager helixZkManager =
- HelixManagerFactory.getZKHelixManager(helixClusterName, brokerId, InstanceType.PARTICIPANT, zkServer);
- final StateMachineEngine stateMachineEngine = helixZkManager.getStateMachineEngine();
- final StateModelFactory<?> stateModelFactory = new EmptyBrokerOnlineOfflineStateModelFactory();
- stateMachineEngine
- .registerStateModelFactory(EmptyBrokerOnlineOfflineStateModelFactory.getStateModelDef(), stateModelFactory);
- helixZkManager.connect();
- if (isSingleTenant) {
- helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, brokerId,
- TagNameUtils.getBrokerTagForTenant(TagNameUtils.DEFAULT_TENANT_NAME));
- } else {
- helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, brokerId, UNTAGGED_BROKER_INSTANCE);
- }
- }
- }
-
- public static void addFakeDataInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- int numInstances)
- throws Exception {
- addFakeDataInstancesToAutoJoinHelixCluster(helixClusterName, zkServer, numInstances, false,
- CommonConstants.Server.DEFAULT_ADMIN_API_PORT);
- }
-
- public static void addFakeDataInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- int numInstances, boolean isSingleTenant)
- throws Exception {
- addFakeDataInstancesToAutoJoinHelixCluster(helixClusterName, zkServer, numInstances, isSingleTenant,
- CommonConstants.Server.DEFAULT_ADMIN_API_PORT);
- }
-
- public static void addFakeDataInstancesToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- int numInstances, boolean isSingleTenant, int adminPort)
- throws Exception {
-
- for (int i = 0; i < numInstances; ++i) {
- final String instanceId = "Server_localhost_" + i;
- addFakeDataInstanceToAutoJoinHelixCluster(helixClusterName, zkServer, instanceId, isSingleTenant, adminPort + i);
- }
- }
-
- public static void addFakeDataInstanceToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- String instanceId)
- throws Exception {
- addFakeDataInstanceToAutoJoinHelixCluster(helixClusterName, zkServer, instanceId, false,
- CommonConstants.Server.DEFAULT_ADMIN_API_PORT);
- }
-
- public static void addFakeDataInstanceToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- String instanceId, boolean isSingleTenant)
- throws Exception {
- addFakeDataInstanceToAutoJoinHelixCluster(helixClusterName, zkServer, instanceId, isSingleTenant,
- CommonConstants.Server.DEFAULT_ADMIN_API_PORT);
- }
-
- public static void addFakeDataInstanceToAutoJoinHelixCluster(String helixClusterName, String zkServer,
- String instanceId, boolean isSingleTenant, int adminPort)
- throws Exception {
- final HelixManager helixZkManager =
- HelixManagerFactory.getZKHelixManager(helixClusterName, instanceId, InstanceType.PARTICIPANT, zkServer);
- final StateMachineEngine stateMachineEngine = helixZkManager.getStateMachineEngine();
- final StateModelFactory<?> stateModelFactory = new EmptySegmentOnlineOfflineStateModelFactory();
- stateMachineEngine
- .registerStateModelFactory(EmptySegmentOnlineOfflineStateModelFactory.getStateModelDef(), stateModelFactory);
- helixZkManager.connect();
- if (isSingleTenant) {
- helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, instanceId,
- TableNameBuilder.OFFLINE.tableNameWithType(TagNameUtils.DEFAULT_TENANT_NAME));
- helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, instanceId,
- TableNameBuilder.REALTIME.tableNameWithType(TagNameUtils.DEFAULT_TENANT_NAME));
- } else {
- helixZkManager.getClusterManagmentTool().addInstanceTag(helixClusterName, instanceId, UNTAGGED_SERVER_INSTANCE);
- }
- HelixConfigScope scope =
- new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.PARTICIPANT, helixClusterName)
- .forParticipant(instanceId).build();
- Map<String, String> props = new HashMap<>();
- props.put(CommonConstants.Helix.Instance.ADMIN_PORT_KEY, String.valueOf(adminPort));
- helixZkManager.getClusterManagmentTool().setConfig(scope, props);
- }
-
- public static String buildBrokerTenantCreateRequestJSON(String tenantName, int numberOfInstances)
- throws JsonProcessingException {
- Tenant tenant = new TenantBuilder(tenantName).setRole(TenantRole.BROKER).setTotalInstances(numberOfInstances)
- .setOfflineInstances(0).setRealtimeInstances(0).build();
- return JsonUtils.objectToString(tenant);
- }
-
- public static String buildServerTenantCreateRequestJSON(String tenantName, int numberOfInstances,
- int offlineInstances, int realtimeInstances)
- throws JsonProcessingException {
- Tenant tenant = new TenantBuilder(tenantName).setRole(TenantRole.SERVER).setTotalInstances(numberOfInstances)
- .setOfflineInstances(offlineInstances).setRealtimeInstances(realtimeInstances).build();
- return JsonUtils.objectToString(tenant);
- }
-}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/EmptyBrokerOnlineOfflineStateModelFactory.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/EmptyBrokerOnlineOfflineStateModelFactory.java
deleted file mode 100644
index b7ae591..0000000
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/EmptyBrokerOnlineOfflineStateModelFactory.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- * 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.controller.helix;
-
-import org.apache.helix.NotificationContext;
-import org.apache.helix.model.Message;
-import org.apache.helix.participant.statemachine.StateModel;
-import org.apache.helix.participant.statemachine.StateModelFactory;
-import org.apache.helix.participant.statemachine.StateModelInfo;
-import org.apache.helix.participant.statemachine.Transition;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class EmptyBrokerOnlineOfflineStateModelFactory extends StateModelFactory<StateModel> {
-
- @Override
- public StateModel createNewStateModel(String partitionName) {
- final EmptyBrokerOnlineOfflineStateModel SegmentOnlineOfflineStateModel = new EmptyBrokerOnlineOfflineStateModel();
- return SegmentOnlineOfflineStateModel;
- }
-
- public EmptyBrokerOnlineOfflineStateModelFactory() {
- }
-
- public static String getStateModelDef() {
- return "BrokerResourceOnlineOfflineStateModel";
- }
-
- @StateModelInfo(states = "{'OFFLINE','ONLINE', 'DROPPED'}", initialState = "OFFLINE")
- public static class EmptyBrokerOnlineOfflineStateModel extends StateModel {
- private static final Logger LOGGER = LoggerFactory.getLogger(EmptyBrokerOnlineOfflineStateModel.class);
-
- @Transition(from = "OFFLINE", to = "ONLINE")
- public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
- LOGGER.info("EmptyBrokerOnlineOfflineStateModel.onBecomeOnlineFromOffline() : " + message);
- }
-
- @Transition(from = "ONLINE", to = "OFFLINE")
- public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
- LOGGER.info("EmptyBrokerOnlineOfflineStateModel.onBecomeOfflineFromOnline() : " + message);
- }
-
- @Transition(from = "OFFLINE", to = "DROPPED")
- public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
- LOGGER.info("EmptyBrokerOnlineOfflineStateModel.onBecomeDroppedFromOffline() : " + message);
- }
-
- @Transition(from = "ONLINE", to = "DROPPED")
- public void onBecomeDroppedFromOnline(Message message, NotificationContext context) {
- LOGGER.info("EmptyBrokerOnlineOfflineStateModel.onBecomeDroppedFromOnline() : " + message);
- }
- }
-}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/EmptySegmentOnlineOfflineStateModelFactory.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/EmptySegmentOnlineOfflineStateModelFactory.java
deleted file mode 100644
index 48b8f05..0000000
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/EmptySegmentOnlineOfflineStateModelFactory.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/**
- * 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.controller.helix;
-
-import org.apache.helix.NotificationContext;
-import org.apache.helix.model.Message;
-import org.apache.helix.participant.statemachine.StateModel;
-import org.apache.helix.participant.statemachine.StateModelFactory;
-import org.apache.helix.participant.statemachine.StateModelInfo;
-import org.apache.helix.participant.statemachine.Transition;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class EmptySegmentOnlineOfflineStateModelFactory extends StateModelFactory<StateModel> {
-
- @Override
- public StateModel createNewStateModel(String partitionName) {
- final EmptySegmentOnlineOfflineStateModel SegmentOnlineOfflineStateModel =
- new EmptySegmentOnlineOfflineStateModel();
- return SegmentOnlineOfflineStateModel;
- }
-
- public EmptySegmentOnlineOfflineStateModelFactory() {
- }
-
- public static String getStateModelDef() {
- return "SegmentOnlineOfflineStateModel";
- }
-
- @StateModelInfo(states = "{'OFFLINE','ONLINE', 'DROPPED'}", initialState = "OFFLINE")
- public static class EmptySegmentOnlineOfflineStateModel extends StateModel {
- private static final Logger LOGGER = LoggerFactory.getLogger(EmptySegmentOnlineOfflineStateModel.class);
-
- @Transition(from = "OFFLINE", to = "ONLINE")
- public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
- LOGGER.info("EmptySegmentOnlineOfflineStateModel.onBecomeOnlineFromOffline() : " + message);
- }
-
- @Transition(from = "OFFLINE", to = "CONSUMING")
- public void onBecomeConsumingFromOffline(Message message, NotificationContext context) {
- LOGGER.info("EmptySegmentOnlineOfflineStateModel.onBecomeConsumingFromOffline() : " + message);
- }
-
- @Transition(from = "ONLINE", to = "OFFLINE")
- public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
- LOGGER.info("EmptySegmentOnlineOfflineStateModel.onBecomeOfflineFromOnline() : " + message);
- }
-
- @Transition(from = "OFFLINE", to = "DROPPED")
- public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
- LOGGER.info("EmptySegmentOnlineOfflineStateModel.onBecomeDroppedFromOffline() : " + message);
- }
-
- @Transition(from = "ONLINE", to = "DROPPED")
- public void onBecomeDroppedFromOnline(Message message, NotificationContext context) {
- LOGGER.info("EmptySegmentOnlineOfflineStateModel.onBecomeDroppedFromOnline() : " + message);
- }
-
- @Transition(from = "CONSUMING", to = "DROPPED")
- public void onBecomeDroppedFromConsuming(Message message, NotificationContext context) {
- LOGGER.info("EmptySegmentOnlineOfflineStateModel.onBecomeDroppedFromConsuming() : " + message);
- }
- }
-}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotFileUploadTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotFileUploadTest.java
index 6d1ba67..d0c7511 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotFileUploadTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotFileUploadTest.java
@@ -25,8 +25,6 @@ import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.ZkStarter;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -41,6 +39,20 @@ import org.testng.annotations.Test;
public class PinotFileUploadTest extends ControllerTest {
private static final String TABLE_NAME = "testTable";
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ startZk();
+ startController();
+ addFakeBrokerInstancesToAutoJoinHelixCluster(5, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(5, true);
+
+ // Adding table
+ TableConfig tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setSegmentAssignmentStrategy("RandomAssignmentStrategy").setNumReplicas(2).build();
+ _helixResourceManager.addTable(tableConfig);
+ }
+
@Test
public void testUploadBogusData()
throws Exception {
@@ -54,28 +66,9 @@ public class PinotFileUploadTest extends ControllerTest {
Assert.assertTrue(statusCode >= 400 && statusCode < 500, "Status code = " + statusCode);
}
- @BeforeClass
- public void setUp()
- throws Exception {
- startZk();
- startController();
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR, 5, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR, 5, true);
-
- Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_BROKER").size(),
- 5);
-
- // Adding table
- TableConfig tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(TABLE_NAME)
- .setSegmentAssignmentStrategy("RandomAssignmentStrategy").setNumReplicas(2).build();
- _helixResourceManager.addTable(tableConfig);
- }
-
@AfterClass
- public void tearDown()
- throws Exception {
+ public void tearDown() {
+ stopFakeInstances();
stopController();
stopZk();
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSegmentRestletResourceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSegmentRestletResourceTest.java
index 65d3d00..95869ec 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSegmentRestletResourceTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotSegmentRestletResourceTest.java
@@ -25,8 +25,6 @@ import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.segment.SegmentMetadata;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.JsonUtils;
-import org.apache.pinot.common.utils.ZkStarter;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.testng.Assert;
@@ -36,34 +34,18 @@ import org.testng.annotations.Test;
public class PinotSegmentRestletResourceTest extends ControllerTest {
- private final static String ZK_SERVER = ZkStarter.DEFAULT_ZK_STR;
- private final static String TABLE_NAME = "testTable";
+ private static final String TABLE_NAME = "testTable";
@BeforeClass
public void setUp()
throws Exception {
startZk();
startController();
+ addFakeBrokerInstancesToAutoJoinHelixCluster(1, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(1, true);
- ControllerRequestBuilderUtil.addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZK_SERVER, 1, true);
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZK_SERVER, 1, true);
-
- while (_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_OFFLINE").size() == 0) {
- Thread.sleep(100);
- }
-
- Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_OFFLINE").size(),
- 1);
- Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_BROKER").size(),
- 1);
- }
-
- @AfterClass
- public void tearDown()
- throws Exception {
- stopController();
- stopZk();
+ Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, "DefaultTenant_OFFLINE").size(), 1);
+ Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, "DefaultTenant_BROKER").size(), 1);
}
@Test
@@ -123,4 +105,11 @@ public class PinotSegmentRestletResourceTest extends ControllerTest {
}
Assert.assertEquals(crcMap.size(), expectedSize);
}
+
+ @AfterClass
+ public void tearDown() {
+ stopFakeInstances();
+ stopController();
+ stopZk();
+ }
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
index 71dd4c0..6e4155d 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTableRestletResourceTest.java
@@ -25,9 +25,7 @@ import org.apache.pinot.common.config.QuotaConfig;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
import org.apache.pinot.common.utils.JsonUtils;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import org.apache.pinot.core.realtime.stream.StreamConfig;
@@ -60,12 +58,8 @@ public class PinotTableRestletResourceTest extends ControllerTest {
startController(config);
_createTableUrl = _controllerRequestURLBuilder.forTableCreate();
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_BROKER_INSTANCES, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_SERVER_INSTANCES, true);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_BROKER_INSTANCES, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_SERVER_INSTANCES, true);
_offlineBuilder.setTableName("testOfflineTable").setTimeColumnName("timeColumn").setTimeType("DAYS")
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("5");
@@ -308,6 +302,7 @@ public class PinotTableRestletResourceTest extends ControllerTest {
@AfterClass
public void tearDown() {
+ stopFakeInstances();
stopController();
stopZk();
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTenantRestletResourceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTenantRestletResourceTest.java
index b5aa4cc..fa38d8a 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTenantRestletResourceTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotTenantRestletResourceTest.java
@@ -23,8 +23,6 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.JsonUtils;
-import org.apache.pinot.common.utils.ZkStarter;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -81,12 +79,8 @@ public class PinotTenantRestletResourceTest extends ControllerTest {
// Add a table to the server
String createTableUrl = _controllerRequestURLBuilder.forTableCreate();
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_BROKER_INSTANCES, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_SERVER_INSTANCES, true);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_BROKER_INSTANCES, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_SERVER_INSTANCES, true);
_offlineBuilder.setTableName("testOfflineTable").setTimeColumnName("timeColumn").setTimeType("DAYS")
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("5").setServerTenant("DefaultTenant");
@@ -101,6 +95,8 @@ public class PinotTenantRestletResourceTest extends ControllerTest {
JsonUtils.stringToJsonNode(sendGetRequest(_controllerRequestURLBuilder.forTablesFromTenant("DefaultTenant")));
assertEquals(tableList.get("tables").size(), 1, "Expected 1 table");
assertEquals(tableList.get("tables").get(0).asText(), "mytable_OFFLINE");
+
+ stopFakeInstances();
}
@AfterClass
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableViewsTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableViewsTest.java
index 0194d42..5ce8ced 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableViewsTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableViewsTest.java
@@ -26,9 +26,7 @@ import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.JsonUtils;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.api.resources.TableViews;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
@@ -52,13 +50,8 @@ public class TableViewsTest extends ControllerTest {
throws Exception {
startZk();
startController();
-
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_BROKER_INSTANCES, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_SERVER_INSTANCES, true);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_BROKER_INSTANCES, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_SERVER_INSTANCES, true);
// Create the offline table and add one segment
TableConfig tableConfig =
@@ -178,6 +171,7 @@ public class TableViewsTest extends ControllerTest {
@AfterClass
public void tearDown() {
+ stopFakeInstances();
stopController();
stopZk();
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerInstanceToggleTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerInstanceToggleTest.java
index 731fa3f..63ec62f 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerInstanceToggleTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerInstanceToggleTest.java
@@ -24,7 +24,6 @@ import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.config.TagNameUtils;
import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.testng.Assert;
@@ -40,17 +39,13 @@ public class ControllerInstanceToggleTest extends ControllerTest {
private static final String BROKER_TAG_NAME = TagNameUtils.getBrokerTagForTenant(null);
private static final int NUM_INSTANCES = 3;
- private final String _helixClusterName = getHelixClusterName();
-
@BeforeClass
- public void setup()
+ public void setUp()
throws Exception {
startZk();
startController();
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, NUM_INSTANCES, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, NUM_INSTANCES, true);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_INSTANCES, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_INSTANCES, true);
}
@Test
@@ -124,6 +119,7 @@ public class ControllerInstanceToggleTest extends ControllerTest {
@AfterClass
public void tearDown() {
+ stopFakeInstances();
stopController();
stopZk();
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerSentinelTestV2.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerSentinelTestV2.java
index 97bf934..68e2dcc 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerSentinelTestV2.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerSentinelTestV2.java
@@ -22,7 +22,6 @@ import java.io.IOException;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.config.TagNameUtils;
import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -31,23 +30,14 @@ import org.testng.annotations.Test;
public class ControllerSentinelTestV2 extends ControllerTest {
- private final String _helixClusterName = getHelixClusterName();
@BeforeClass
- public void setup()
+ public void setUp()
throws Exception {
startZk();
startController();
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, 20, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, 20, true);
- }
-
- @AfterClass
- public void tearDown() {
- stopController();
- stopZk();
+ addFakeBrokerInstancesToAutoJoinHelixCluster(20, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(20, true);
}
@Test
@@ -91,4 +81,11 @@ public class ControllerSentinelTestV2 extends ControllerTest {
Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(_helixClusterName,
TagNameUtils.getOfflineTagForTenant(TagNameUtils.DEFAULT_TENANT_NAME)).size(), 20);
}
+
+ @AfterClass
+ public void tearDown() {
+ stopFakeInstances();
+ stopController();
+ stopZk();
+ }
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTenantTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTenantTest.java
index 59923d5..ac7d4fb 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTenantTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTenantTest.java
@@ -24,7 +24,6 @@ import org.apache.pinot.common.config.TagNameUtils;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.JsonUtils;
-import org.apache.pinot.common.utils.ZkStarter;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -42,18 +41,14 @@ public class ControllerTenantTest extends ControllerTest {
private static final int NUM_REALTIME_SERVERS_PER_TAG = 1;
private static final int NUM_SERVERS_PER_TAG = NUM_OFFLINE_SERVERS_PER_TAG + NUM_REALTIME_SERVERS_PER_TAG;
- private final String _helixClusterName = getHelixClusterName();
-
@BeforeClass
public void setUp()
throws Exception {
startZk();
startController();
ZKMetadataProvider.setClusterTenantIsolationEnabled(_propertyStore, false);
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, NUM_INSTANCES);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, NUM_INSTANCES);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_INSTANCES, false);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_INSTANCES, false);
}
@Test
@@ -62,9 +57,7 @@ public class ControllerTenantTest extends ControllerTest {
// Create broker tenants
for (int i = 1; i <= NUM_BROKER_TAGS; i++) {
String brokerTenant = BROKER_TAG_PREFIX + i;
- String payload =
- ControllerRequestBuilderUtil.buildBrokerTenantCreateRequestJSON(brokerTenant, NUM_BROKERS_PER_TAG);
- sendPostRequest(_controllerRequestURLBuilder.forTenantCreate(), payload);
+ createBrokerTenant(brokerTenant, NUM_BROKERS_PER_TAG);
Assert.assertEquals(
_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, TagNameUtils.getBrokerTagForTenant(brokerTenant))
.size(), NUM_BROKERS_PER_TAG);
@@ -90,8 +83,7 @@ public class ControllerTenantTest extends ControllerTest {
// Update broker tenants
for (int i = 0; i <= NUM_INSTANCES - (NUM_BROKER_TAGS - 1) * NUM_BROKERS_PER_TAG; i++) {
String brokerTenant = BROKER_TAG_PREFIX + 1;
- String payload = ControllerRequestBuilderUtil.buildBrokerTenantCreateRequestJSON(brokerTenant, i);
- sendPutRequest(_controllerRequestURLBuilder.forTenantCreate(), payload);
+ updateBrokerTenant(brokerTenant, i);
Assert.assertEquals(
_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, TagNameUtils.getBrokerTagForTenant(brokerTenant))
.size(), i);
@@ -119,10 +111,7 @@ public class ControllerTenantTest extends ControllerTest {
// Create server tenants
for (int i = 1; i <= NUM_SERVER_TAGS; i++) {
String serverTenant = SERVER_TAG_PREFIX + i;
- String payload = ControllerRequestBuilderUtil
- .buildServerTenantCreateRequestJSON(serverTenant, NUM_SERVERS_PER_TAG, NUM_OFFLINE_SERVERS_PER_TAG,
- NUM_REALTIME_SERVERS_PER_TAG);
- sendPostRequest(_controllerRequestURLBuilder.forTenantCreate(), payload);
+ createServerTenant(serverTenant, NUM_OFFLINE_SERVERS_PER_TAG, NUM_REALTIME_SERVERS_PER_TAG);
Assert.assertEquals(
_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, TagNameUtils.getOfflineTagForTenant(serverTenant))
.size(), NUM_OFFLINE_SERVERS_PER_TAG);
@@ -152,10 +141,7 @@ public class ControllerTenantTest extends ControllerTest {
// Note: server tenants cannot scale down
for (int i = 0; i <= (NUM_INSTANCES - NUM_SERVER_TAGS * NUM_SERVERS_PER_TAG) / 2; i++) {
String serverTenant = SERVER_TAG_PREFIX + 1;
- String payload = ControllerRequestBuilderUtil
- .buildServerTenantCreateRequestJSON(serverTenant, NUM_SERVERS_PER_TAG + i * 2,
- NUM_OFFLINE_SERVERS_PER_TAG + i, NUM_REALTIME_SERVERS_PER_TAG + i);
- sendPutRequest(_controllerRequestURLBuilder.forTenantCreate(), payload);
+ updateServerTenant(serverTenant, NUM_OFFLINE_SERVERS_PER_TAG + i, NUM_REALTIME_SERVERS_PER_TAG + i);
Assert.assertEquals(
_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, TagNameUtils.getOfflineTagForTenant(serverTenant))
.size(), NUM_OFFLINE_SERVERS_PER_TAG + i);
@@ -185,6 +171,7 @@ public class ControllerTenantTest extends ControllerTest {
@AfterClass
public void tearDown() {
+ stopFakeInstances();
stopController();
stopZk();
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
index 4d49192..be2dce7 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.helix;
+import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.base.Preconditions;
import java.io.BufferedReader;
import java.io.BufferedWriter;
@@ -28,7 +29,10 @@ import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.HttpURLConnection;
import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.PutMethod;
@@ -39,21 +43,40 @@ import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.NotificationContext;
import org.apache.helix.ZNRecord;
import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.HelixConfigScope;
+import org.apache.helix.model.Message;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.pinot.common.config.TagNameUtils;
+import org.apache.pinot.common.config.Tenant;
import org.apache.pinot.common.data.DimensionFieldSpec;
import org.apache.pinot.common.data.FieldSpec;
import org.apache.pinot.common.data.MetricFieldSpec;
import org.apache.pinot.common.data.Schema;
+import org.apache.pinot.common.utils.JsonUtils;
+import org.apache.pinot.common.utils.TenantRole;
import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerStarter;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.testng.Assert;
+import static org.apache.pinot.common.utils.CommonConstants.Helix.Instance.ADMIN_PORT_KEY;
+import static org.apache.pinot.common.utils.CommonConstants.Helix.UNTAGGED_BROKER_INSTANCE;
+import static org.apache.pinot.common.utils.CommonConstants.Helix.UNTAGGED_SERVER_INSTANCE;
+import static org.apache.pinot.common.utils.CommonConstants.Server.DEFAULT_ADMIN_API_PORT;
+
/**
* Base class for controller tests.
@@ -63,6 +86,11 @@ public abstract class ControllerTest {
protected static final int DEFAULT_CONTROLLER_PORT = 18998;
protected static final String DEFAULT_DATA_DIR =
new File(FileUtils.getTempDirectoryPath(), "test-controller-" + System.currentTimeMillis()).getAbsolutePath();
+ protected static final String BROKER_INSTANCE_ID_PREFIX = "Broker_localhost_";
+ protected static final String SERVER_INSTANCE_ID_PREFIX = "Server_localhost_";
+
+ protected final String _helixClusterName = getHelixClusterName();
+ protected final List<HelixManager> _fakeInstanceHelixManagers = new ArrayList<>();
protected int _controllerPort;
protected String _controllerBaseApiUrl;
@@ -103,7 +131,7 @@ public abstract class ControllerTest {
config.setControllerPort(Integer.toString(DEFAULT_CONTROLLER_PORT));
config.setDataDir(DEFAULT_DATA_DIR);
config.setZkStr(ZkStarter.DEFAULT_ZK_STR);
- config.setHelixClusterName(getHelixClusterName());
+ config.setHelixClusterName(_helixClusterName);
return config;
}
@@ -120,7 +148,10 @@ public abstract class ControllerTest {
_controllerRequestURLBuilder = ControllerRequestURLBuilder.baseUrl(_controllerBaseApiUrl);
_controllerDataDir = config.getDataDir();
- startControllerStarter(config);
+ _controllerStarter = getControllerStarter(config);
+ _controllerStarter.start();
+ _helixResourceManager = _controllerStarter.getHelixResourceManager();
+ _helixManager = _controllerStarter.getHelixControllerManager();
// HelixResourceManager is null in Helix only mode, while HelixManager is null in Pinot only mode.
switch (_controllerStarter.getControllerMode()) {
@@ -132,7 +163,7 @@ public abstract class ControllerTest {
// TODO: Enable periodic rebalance per 10 seconds as a temporary work-around for the Helix issue:
// https://github.com/apache/helix/issues/331. Remove this after Helix fixing the issue.
_helixAdmin.setConfig(
- new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(getHelixClusterName())
+ new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(_helixClusterName)
.build(),
Collections.singletonMap(ClusterConfig.ClusterConfigProperty.REBALANCE_TIMER_PERIOD.name(), "10000"));
break;
@@ -143,27 +174,205 @@ public abstract class ControllerTest {
}
}
- protected void startControllerStarter(ControllerConf config) {
- _controllerStarter = getControllerStarter(config);
- _controllerStarter.start();
- _helixResourceManager = _controllerStarter.getHelixResourceManager();
- _helixManager = _controllerStarter.getHelixControllerManager();
- }
-
protected ControllerStarter getControllerStarter(ControllerConf config) {
return new ControllerStarter(config);
}
protected void stopController() {
- stopControllerStarter();
+ _controllerStarter.stop();
+ _controllerStarter = null;
FileUtils.deleteQuietly(new File(_controllerDataDir));
}
- protected void stopControllerStarter() {
- Assert.assertNotNull(_controllerStarter);
+ protected void addFakeBrokerInstancesToAutoJoinHelixCluster(int numInstances, boolean isSingleTenant)
+ throws Exception {
+ for (int i = 0; i < numInstances; i++) {
+ addFakeBrokerInstanceToAutoJoinHelixCluster(BROKER_INSTANCE_ID_PREFIX + i, isSingleTenant);
+ }
+ }
- _controllerStarter.stop();
- _controllerStarter = null;
+ protected void addFakeBrokerInstanceToAutoJoinHelixCluster(String instanceId, boolean isSingleTenant)
+ throws Exception {
+ HelixManager helixManager = HelixManagerFactory
+ .getZKHelixManager(_helixClusterName, instanceId, InstanceType.PARTICIPANT, ZkStarter.DEFAULT_ZK_STR);
+ helixManager.getStateMachineEngine()
+ .registerStateModelFactory(FakeBrokerResourceOnlineOfflineStateModelFactory.STATE_MODEL_DEF,
+ FakeBrokerResourceOnlineOfflineStateModelFactory.FACTORY_INSTANCE);
+ helixManager.connect();
+ HelixAdmin helixAdmin = helixManager.getClusterManagmentTool();
+ if (isSingleTenant) {
+ helixAdmin.addInstanceTag(_helixClusterName, instanceId, TagNameUtils.getBrokerTagForTenant(null));
+ } else {
+ helixAdmin.addInstanceTag(_helixClusterName, instanceId, UNTAGGED_BROKER_INSTANCE);
+ }
+ _fakeInstanceHelixManagers.add(helixManager);
+ }
+
+ public static class FakeBrokerResourceOnlineOfflineStateModelFactory extends StateModelFactory<StateModel> {
+ private static final String STATE_MODEL_DEF = "BrokerResourceOnlineOfflineStateModel";
+ private static final FakeBrokerResourceOnlineOfflineStateModelFactory FACTORY_INSTANCE =
+ new FakeBrokerResourceOnlineOfflineStateModelFactory();
+ private static final FakeBrokerResourceOnlineOfflineStateModel STATE_MODEL_INSTANCE =
+ new FakeBrokerResourceOnlineOfflineStateModel();
+
+ private FakeBrokerResourceOnlineOfflineStateModelFactory() {
+ }
+
+ @Override
+ public StateModel createNewStateModel(String resourceName, String partitionName) {
+ return STATE_MODEL_INSTANCE;
+ }
+
+ @SuppressWarnings("unused")
+ @StateModelInfo(states = "{'OFFLINE', 'ONLINE', 'DROPPED'}", initialState = "OFFLINE")
+ public static class FakeBrokerResourceOnlineOfflineStateModel extends StateModel {
+ private static final Logger LOGGER = LoggerFactory.getLogger(FakeBrokerResourceOnlineOfflineStateModel.class);
+
+ private FakeBrokerResourceOnlineOfflineStateModel() {
+ }
+
+ @Transition(from = "OFFLINE", to = "ONLINE")
+ public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOnlineFromOffline(): {}", message);
+ }
+
+ @Transition(from = "OFFLINE", to = "DROPPED")
+ public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeDroppedFromOffline(): {}", message);
+ }
+
+ @Transition(from = "ONLINE", to = "OFFLINE")
+ public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOfflineFromOnline(): {}", message);
+ }
+
+ @Transition(from = "ONLINE", to = "DROPPED")
+ public void onBecomeDroppedFromOnline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeDroppedFromOnline(): {}", message);
+ }
+
+ @Transition(from = "ERROR", to = "OFFLINE")
+ public void onBecomeOfflineFromError(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOfflineFromError(): {}", message);
+ }
+ }
+ }
+
+ protected void addFakeServerInstancesToAutoJoinHelixCluster(int numInstances, boolean isSingleTenant)
+ throws Exception {
+ addFakeServerInstancesToAutoJoinHelixCluster(numInstances, isSingleTenant, DEFAULT_ADMIN_API_PORT);
+ }
+
+ protected void addFakeServerInstancesToAutoJoinHelixCluster(int numInstances, boolean isSingleTenant,
+ int baseAdminPort)
+ throws Exception {
+ for (int i = 0; i < numInstances; i++) {
+ addFakeServerInstanceToAutoJoinHelixCluster(SERVER_INSTANCE_ID_PREFIX + i, isSingleTenant, baseAdminPort + i);
+ }
+ }
+
+ protected void addFakeServerInstanceToAutoJoinHelixCluster(String instanceId, boolean isSingleTenant)
+ throws Exception {
+ addFakeServerInstanceToAutoJoinHelixCluster(instanceId, isSingleTenant, DEFAULT_ADMIN_API_PORT);
+ }
+
+ protected void addFakeServerInstanceToAutoJoinHelixCluster(String instanceId, boolean isSingleTenant, int adminPort)
+ throws Exception {
+ HelixManager helixManager = HelixManagerFactory
+ .getZKHelixManager(_helixClusterName, instanceId, InstanceType.PARTICIPANT, ZkStarter.DEFAULT_ZK_STR);
+ helixManager.getStateMachineEngine()
+ .registerStateModelFactory(FakeSegmentOnlineOfflineStateModelFactory.STATE_MODEL_DEF,
+ FakeSegmentOnlineOfflineStateModelFactory.FACTORY_INSTANCE);
+ helixManager.connect();
+ HelixAdmin helixAdmin = helixManager.getClusterManagmentTool();
+ if (isSingleTenant) {
+ helixAdmin.addInstanceTag(_helixClusterName, instanceId, TagNameUtils.getOfflineTagForTenant(null));
+ helixAdmin.addInstanceTag(_helixClusterName, instanceId, TagNameUtils.getRealtimeTagForTenant(null));
+ } else {
+ helixAdmin.addInstanceTag(_helixClusterName, instanceId, UNTAGGED_SERVER_INSTANCE);
+ }
+ HelixConfigScope configScope =
+ new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.PARTICIPANT, _helixClusterName)
+ .forParticipant(instanceId).build();
+ helixAdmin.setConfig(configScope, Collections.singletonMap(ADMIN_PORT_KEY, Integer.toString(adminPort)));
+ _fakeInstanceHelixManagers.add(helixManager);
+ }
+
+ public static class FakeSegmentOnlineOfflineStateModelFactory extends StateModelFactory<StateModel> {
+ private static final String STATE_MODEL_DEF = "SegmentOnlineOfflineStateModel";
+ private static final FakeSegmentOnlineOfflineStateModelFactory FACTORY_INSTANCE =
+ new FakeSegmentOnlineOfflineStateModelFactory();
+ private static final FakeSegmentOnlineOfflineStateModel STATE_MODEL_INSTANCE =
+ new FakeSegmentOnlineOfflineStateModel();
+
+ private FakeSegmentOnlineOfflineStateModelFactory() {
+ }
+
+ @Override
+ public StateModel createNewStateModel(String resourceName, String partitionName) {
+ return STATE_MODEL_INSTANCE;
+ }
+
+ @SuppressWarnings("unused")
+ @StateModelInfo(states = "{'OFFLINE', 'ONLINE', 'CONSUMING', 'DROPPED'}", initialState = "OFFLINE")
+ public static class FakeSegmentOnlineOfflineStateModel extends StateModel {
+ private static final Logger LOGGER = LoggerFactory.getLogger(FakeSegmentOnlineOfflineStateModel.class);
+
+ private FakeSegmentOnlineOfflineStateModel() {
+ }
+
+ @Transition(from = "OFFLINE", to = "ONLINE")
+ public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOnlineFromOffline(): {}", message);
+ }
+
+ @Transition(from = "OFFLINE", to = "CONSUMING")
+ public void onBecomeConsumingFromOffline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeConsumingFromOffline(): {}", message);
+ }
+
+ @Transition(from = "OFFLINE", to = "DROPPED")
+ public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeDroppedFromOffline(): {}", message);
+ }
+
+ @Transition(from = "ONLINE", to = "OFFLINE")
+ public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOfflineFromOnline(): {}", message);
+ }
+
+ @Transition(from = "ONLINE", to = "DROPPED")
+ public void onBecomeDroppedFromOnline(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeDroppedFromOnline(): {}", message);
+ }
+
+ @Transition(from = "CONSUMING", to = "OFFLINE")
+ public void onBecomeOfflineFromConsuming(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOfflineFromConsuming(): {}", message);
+ }
+
+ @Transition(from = "CONSUMING", to = "ONLINE")
+ public void onBecomeOnlineFromConsuming(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOnlineFromConsuming(): {}", message);
+ }
+
+ @Transition(from = "CONSUMING", to = "DROPPED")
+ public void onBecomeDroppedFromConsuming(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeDroppedFromConsuming(): {}", message);
+ }
+
+ @Transition(from = "ERROR", to = "OFFLINE")
+ public void onBecomeOfflineFromError(Message message, NotificationContext context) {
+ LOGGER.debug("onBecomeOfflineFromError(): {}", message);
+ }
+ }
+ }
+
+ protected void stopFakeInstances() {
+ for (HelixManager helixManager : _fakeInstanceHelixManagers) {
+ helixManager.disconnect();
+ }
+ _fakeInstanceHelixManagers.clear();
}
protected Schema createDummySchema(String tableName) {
@@ -193,6 +402,45 @@ public abstract class ControllerTest {
Assert.assertEquals(postMethod.getStatusCode(), 200);
}
+ protected String getBrokerTenantRequestPayload(String tenantName, int numBrokers)
+ throws JsonProcessingException {
+ Tenant tenant =
+ new Tenant.TenantBuilder(tenantName).setRole(TenantRole.BROKER).setTotalInstances(numBrokers).build();
+ return JsonUtils.objectToString(tenant);
+ }
+
+ protected void createBrokerTenant(String tenantName, int numBrokers)
+ throws IOException {
+ sendPostRequest(_controllerRequestURLBuilder.forTenantCreate(),
+ getBrokerTenantRequestPayload(tenantName, numBrokers));
+ }
+
+ protected void updateBrokerTenant(String tenantName, int numBrokers)
+ throws IOException {
+ sendPutRequest(_controllerRequestURLBuilder.forTenantCreate(),
+ getBrokerTenantRequestPayload(tenantName, numBrokers));
+ }
+
+ protected String getServerTenantRequestPayload(String tenantName, int numOfflineServers, int numRealtimeServers)
+ throws JsonProcessingException {
+ Tenant tenant = new Tenant.TenantBuilder(tenantName).setRole(TenantRole.SERVER)
+ .setTotalInstances(numOfflineServers + numRealtimeServers).setOfflineInstances(numOfflineServers)
+ .setRealtimeInstances(numRealtimeServers).build();
+ return JsonUtils.objectToString(tenant);
+ }
+
+ protected void createServerTenant(String tenantName, int numOfflineServers, int numRealtimeServers)
+ throws IOException {
+ sendPostRequest(_controllerRequestURLBuilder.forTenantCreate(),
+ getServerTenantRequestPayload(tenantName, numOfflineServers, numRealtimeServers));
+ }
+
+ protected void updateServerTenant(String tenantName, int numOfflineServers, int numRealtimeServers)
+ throws IOException {
+ sendPutRequest(_controllerRequestURLBuilder.forTenantCreate(),
+ getServerTenantRequestPayload(tenantName, numOfflineServers, numRealtimeServers));
+ }
+
public static String sendGetRequest(String urlString)
throws IOException {
return constructResponse(new URL(urlString).openStream());
@@ -211,7 +459,7 @@ public abstract class ControllerTest {
if (payload != null && !payload.isEmpty()) {
httpConnection.setDoOutput(true);
try (BufferedWriter writer = new BufferedWriter(
- new OutputStreamWriter(httpConnection.getOutputStream(), "UTF-8"))) {
+ new OutputStreamWriter(httpConnection.getOutputStream(), StandardCharsets.UTF_8))) {
writer.write(payload, 0, payload.length());
writer.flush();
}
@@ -226,8 +474,8 @@ public abstract class ControllerTest {
httpConnection.setDoOutput(true);
httpConnection.setRequestMethod("PUT");
- try (
- BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(httpConnection.getOutputStream(), "UTF-8"))) {
+ try (BufferedWriter writer = new BufferedWriter(
+ new OutputStreamWriter(httpConnection.getOutputStream(), StandardCharsets.UTF_8))) {
writer.write(payload);
writer.flush();
}
@@ -246,7 +494,7 @@ public abstract class ControllerTest {
private static String constructResponse(InputStream inputStream)
throws IOException {
- try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, "UTF-8"))) {
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
StringBuilder responseBuilder = new StringBuilder();
String line;
while ((line = reader.readLine()) != null) {
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/HelixHelperTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/HelixHelperTest.java
index 191241d..4f2d319 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/HelixHelperTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/HelixHelperTest.java
@@ -35,7 +35,6 @@ import org.testng.annotations.Test;
public class HelixHelperTest extends ControllerTest {
public static final String RESOURCE_NAME = "potato_OFFLINE";
public static final String INSTANCE_NAME = "Server_1.2.3.4_1234";
- private String helixClusterName;
@BeforeClass
public void setUp() {
@@ -46,8 +45,7 @@ public class HelixHelperTest extends ControllerTest {
idealState.setStateModelDefRef("OnlineOffline");
idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
idealState.setReplicas("0");
- helixClusterName = getHelixClusterName();
- _helixAdmin.addResource(helixClusterName, RESOURCE_NAME, idealState);
+ _helixAdmin.addResource(_helixClusterName, RESOURCE_NAME, idealState);
}
/**
@@ -68,7 +66,7 @@ public class HelixHelperTest extends ControllerTest {
}
}, RetryPolicies.noDelayRetryPolicy(1));
- IdealState resourceIdealState = _helixAdmin.getResourceIdealState(helixClusterName, RESOURCE_NAME);
+ IdealState resourceIdealState = _helixAdmin.getResourceIdealState(_helixClusterName, RESOURCE_NAME);
for (int i = 0; i < numSegments; i++) {
Assert.assertEquals(resourceIdealState.getInstanceStateMap("segment_" + i).get(INSTANCE_NAME), "ONLINE");
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java
index 14461be..884909f 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java
@@ -72,12 +72,12 @@ public class PinotControllerModeTest extends ControllerTest {
// There should be no partition in the external view because the resource is disabled
TestUtils.waitForCondition(aVoid -> {
ExternalView leadControllerResourceExternalView =
- helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
+ helixAdmin.getResourceExternalView(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME);
return leadControllerResourceExternalView.getPartitionSet().isEmpty();
}, TIMEOUT_IN_MS, "There should be no partition in the disabled resource's external view");
// Enable the lead controller resource, and the first controller should be the MASTER for all partitions
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
+ helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, true);
checkInstanceState(helixAdmin, "MASTER");
// Start the second dual-mode controller
@@ -94,11 +94,11 @@ public class PinotControllerModeTest extends ControllerTest {
checkInstanceState(helixAdmin, "MASTER");
// Disable the lead controller resource, and there should be only one OFFLINE instance for each partition
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, false);
+ helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, false);
checkInstanceState(helixAdmin, "OFFLINE");
// Re-enable the lead controller resource, and there should be only one MASTER instance for each partition
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
+ helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, true);
checkInstanceState(helixAdmin, "MASTER");
// Stop the second controller, and there should still be only one MASTER instance for each partition
@@ -138,7 +138,7 @@ public class PinotControllerModeTest extends ControllerTest {
// Enabling the lead controller resource before setting up the Pinot cluster should fail
try {
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
+ helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, true);
Assert.fail("Enabling the lead controller resource before setting up the Pinot cluster should fail");
} catch (Exception e) {
// Expected
@@ -153,12 +153,12 @@ public class PinotControllerModeTest extends ControllerTest {
// There should be no partition in the external view because the resource is disabled
TestUtils.waitForCondition(aVoid -> {
ExternalView leadControllerResourceExternalView =
- helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
+ helixAdmin.getResourceExternalView(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME);
return leadControllerResourceExternalView.getPartitionSet().isEmpty();
}, TIMEOUT_IN_MS, "There should be no partition in the disabled resource's external view");
// Enable the lead controller resource, and the first Pinot-only controller should be the MASTER for all partitions
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
+ helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, true);
checkInstanceState(helixAdmin, "MASTER");
// Start the second Pinot-only controller
@@ -175,11 +175,11 @@ public class PinotControllerModeTest extends ControllerTest {
checkInstanceState(helixAdmin, "MASTER");
// Disable the lead controller resource, and there should be only one OFFLINE instance for each partition
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, false);
+ helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, false);
checkInstanceState(helixAdmin, "OFFLINE");
// Re-enable the lead controller resource, and there should be only one MASTER instance for each partition
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
+ helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, true);
checkInstanceState(helixAdmin, "MASTER");
// Stop the second Pinot-only controller, and there should still be only one MASTER instance for each partition
@@ -190,7 +190,7 @@ public class PinotControllerModeTest extends ControllerTest {
firstPinotOnlyController.stop();
TestUtils.waitForCondition(aVoid -> {
ExternalView leadControllerResourceExternalView =
- helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
+ helixAdmin.getResourceExternalView(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME);
return leadControllerResourceExternalView.getPartitionSet().isEmpty();
}, TIMEOUT_IN_MS, "Without live instance, there should be no partition in the external view");
@@ -201,7 +201,7 @@ public class PinotControllerModeTest extends ControllerTest {
private void checkInstanceState(HelixAdmin helixAdmin, String expectedInstanceState) {
TestUtils.waitForCondition(aVoid -> {
ExternalView leadControllerResourceExternalView =
- helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
+ helixAdmin.getResourceExternalView(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME);
Set<String> partitionSet = leadControllerResourceExternalView.getPartitionSet();
if (partitionSet.size() != NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE) {
return false;
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java
index d65e818..8afa55c 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotResourceManagerTest.java
@@ -22,13 +22,11 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-import org.apache.helix.manager.zk.ZkClient;
import org.apache.helix.model.IdealState;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.metadata.segment.OfflineSegmentZKMetadata;
import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -37,28 +35,20 @@ import org.testng.annotations.Test;
public class PinotResourceManagerTest extends ControllerTest {
- private final static String TABLE_NAME = "testTable";
-
- private ZkStarter.ZookeeperInstance _zookeeperInstance;
- private ZkClient _zkClient;
+ private static final String TABLE_NAME = "testTable";
@BeforeClass
public void setUp()
throws Exception {
- _zookeeperInstance = ZkStarter.startLocalZkServer();
- _zkClient = new ZkClient(ZkStarter.DEFAULT_ZK_STR);
-
+ startZk();
startController();
+ addFakeBrokerInstancesToAutoJoinHelixCluster(1, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(1, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR, 1, true);
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR, 1, true);
- Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_BROKER").size(), 1);
- Assert
- .assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_OFFLINE").size(), 1);
+ Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, "DefaultTenant_BROKER").size(), 1);
+ Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, "DefaultTenant_OFFLINE").size(), 1);
Assert
- .assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_REALTIME").size(), 1);
+ .assertEquals(_helixAdmin.getInstancesInClusterWithTag(_helixClusterName, "DefaultTenant_REALTIME").size(), 1);
// Adding table
TableConfig tableConfig =
@@ -78,11 +68,13 @@ public class PinotResourceManagerTest extends ControllerTest {
Assert.assertTrue(_helixResourceManager.updateZkMetadata("testTable_OFFLINE", segmentZKMetadata));
// Update ZK metadata
- Assert.assertEquals(
- _helixResourceManager.getSegmentMetadataZnRecord("testTable_OFFLINE", "testSegment").getVersion(), 0);
+ Assert
+ .assertEquals(_helixResourceManager.getSegmentMetadataZnRecord("testTable_OFFLINE", "testSegment").getVersion(),
+ 0);
Assert.assertTrue(_helixResourceManager.updateZkMetadata("testTable_OFFLINE", segmentZKMetadata, 0));
- Assert.assertEquals(
- _helixResourceManager.getSegmentMetadataZnRecord("testTable_OFFLINE", "testSegment").getVersion(), 1);
+ Assert
+ .assertEquals(_helixResourceManager.getSegmentMetadataZnRecord("testTable_OFFLINE", "testSegment").getVersion(),
+ 1);
Assert.assertFalse(_helixResourceManager.updateZkMetadata("testTable_OFFLINE", segmentZKMetadata, 0));
}
@@ -104,14 +96,14 @@ public class PinotResourceManagerTest extends ControllerTest {
_helixResourceManager
.addNewSegment(TABLE_NAME, SegmentMetadataMockUtils.mockSegmentMetadata(TABLE_NAME), "downloadUrl");
}
- IdealState idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), offlineTableName);
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, offlineTableName);
Set<String> segments = idealState.getPartitionSet();
Assert.assertEquals(segments.size(), 2);
for (String segmentName : segments) {
_helixResourceManager.deleteSegment(offlineTableName, segmentName);
}
- idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), offlineTableName);
+ idealState = _helixAdmin.getResourceIdealState(_helixClusterName, offlineTableName);
Assert.assertEquals(idealState.getPartitionSet().size(), 0);
// Concurrent segment deletion
@@ -130,7 +122,7 @@ public class PinotResourceManagerTest extends ControllerTest {
addSegmentExecutor.shutdown();
addSegmentExecutor.awaitTermination(1, TimeUnit.MINUTES);
- idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), offlineTableName);
+ idealState = _helixAdmin.getResourceIdealState(_helixClusterName, offlineTableName);
Assert.assertEquals(idealState.getPartitionSet().size(), 30);
ExecutorService deleteSegmentExecutor = Executors.newFixedThreadPool(3);
@@ -145,14 +137,14 @@ public class PinotResourceManagerTest extends ControllerTest {
deleteSegmentExecutor.shutdown();
deleteSegmentExecutor.awaitTermination(1, TimeUnit.MINUTES);
- idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), offlineTableName);
+ idealState = _helixAdmin.getResourceIdealState(_helixClusterName, offlineTableName);
Assert.assertEquals(idealState.getPartitionSet().size(), 0);
}
@AfterClass
public void tearDown() {
- _helixResourceManager.stop();
- _zkClient.close();
- ZkStarter.stopLocalZkServer(_zookeeperInstance);
+ stopFakeInstances();
+ startController();
+ stopZk();
}
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java
index fc07811..d2e0753 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java
@@ -51,9 +51,7 @@ import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
import org.apache.pinot.common.utils.TenantRole;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.util.TestUtils;
import org.testng.Assert;
@@ -82,8 +80,6 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
private static final int MAXIMUM_NUMBER_OF_CONTROLLER_INSTANCES = 10;
private static final long TIMEOUT_IN_MS = 10_000L;
- private final String _helixClusterName = getHelixClusterName();
-
@BeforeClass
public void setUp()
throws Exception {
@@ -91,13 +87,8 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
ControllerConf config = getDefaultControllerConfiguration();
config.setTenantIsolationEnabled(false);
startController(config);
-
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, NUM_INSTANCES,
- false);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(_helixClusterName, ZkStarter.DEFAULT_ZK_STR, NUM_INSTANCES, false,
- BASE_SERVER_ADMIN_PORT);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_INSTANCES, false);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_INSTANCES, false, BASE_SERVER_ADMIN_PORT);
// Create server tenant on all Servers
Tenant serverTenant =
@@ -105,7 +96,7 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
.build();
_helixResourceManager.createServerTenant(serverTenant);
- _helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
+ _helixAdmin.enableResource(_helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, true);
}
@Test
@@ -114,7 +105,7 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
Set<String> servers = _helixResourceManager.getAllInstancesForServerTenant(SERVER_TENANT_NAME);
BiMap<String, String> endpoints = _helixResourceManager.getDataInstanceAdminEndpoints(servers);
for (int i = 0; i < NUM_INSTANCES; i++) {
- Assert.assertTrue(endpoints.inverse().containsKey("localhost:" + String.valueOf(BASE_SERVER_ADMIN_PORT + i)));
+ Assert.assertTrue(endpoints.inverse().containsKey("localhost:" + (BASE_SERVER_ADMIN_PORT + i)));
}
}
@@ -172,7 +163,7 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
private void addAndRemoveNewInstanceConfig(ZkClient zkClient)
throws Exception {
int biggerRandomNumber = NUM_INSTANCES + new Random().nextInt(NUM_INSTANCES);
- String instanceName = "Server_localhost_" + String.valueOf(biggerRandomNumber);
+ String instanceName = "Server_localhost_" + biggerRandomNumber;
String instanceConfigPath = PropertyPathBuilder.instanceConfig(_helixClusterName, instanceName);
Assert.assertFalse(zkClient.exists(instanceConfigPath));
List<String> instances = _helixResourceManager.getAllInstances();
@@ -473,7 +464,7 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
@Test
public void testLeadControllerResource() {
IdealState leadControllerResourceIdealState = _helixResourceManager.getHelixAdmin()
- .getResourceIdealState(getHelixClusterName(), CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME);
+ .getResourceIdealState(_helixClusterName, CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME);
Assert.assertTrue(leadControllerResourceIdealState.isValid());
Assert.assertTrue(leadControllerResourceIdealState.isEnabled());
Assert.assertEquals(leadControllerResourceIdealState.getInstanceGroupTag(),
@@ -488,7 +479,7 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
TestUtils.waitForCondition(aVoid -> {
ExternalView leadControllerResourceExternalView = _helixResourceManager.getHelixAdmin()
- .getResourceExternalView(getHelixClusterName(), CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME);
+ .getResourceExternalView(_helixClusterName, CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME);
for (String partition : leadControllerResourceExternalView.getPartitionSet()) {
Map<String, String> stateMap = leadControllerResourceExternalView.getStateMap(partition);
Map.Entry<String, String> entry = stateMap.entrySet().iterator().next();
@@ -527,7 +518,7 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
crushEdRebalanceStrategy.init(LEAD_CONTROLLER_RESOURCE_NAME, partitions, states, Integer.MAX_VALUE);
ClusterDataCache clusterDataCache = new ClusterDataCache();
- PropertyKey.Builder keyBuilder = new PropertyKey.Builder(getHelixClusterName());
+ PropertyKey.Builder keyBuilder = new PropertyKey.Builder(_helixClusterName);
HelixDataAccessor accessor = _helixManager.getHelixDataAccessor();
ClusterConfig clusterConfig = accessor.getProperty(keyBuilder.clusterConfig());
clusterDataCache.setClusterConfig(clusterConfig);
@@ -580,6 +571,7 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
@AfterClass
public void tearDown() {
+ stopFakeInstances();
stopController();
stopZk();
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/ReplicaGroupRebalanceStrategyTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/ReplicaGroupRebalanceStrategyTest.java
index 84c8b50..ec5bda1 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/ReplicaGroupRebalanceStrategyTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/ReplicaGroupRebalanceStrategyTest.java
@@ -31,9 +31,7 @@ import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.partition.ReplicaGroupPartitionAssignment;
import org.apache.pinot.common.partition.ReplicaGroupPartitionAssignmentGenerator;
import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.controller.utils.ReplicaGroupTestUtils;
import org.testng.Assert;
@@ -49,9 +47,9 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
private static final int INITIAL_NUM_SEGMENTS = 20;
private static final String TABLE_NAME = "testReplicaRebalanceReplace";
- private final static String PARTITION_COLUMN = "memberId";
- private final static String OFFLINE_TENENT_NAME = "DefaultTenant_OFFLINE";
- private final static String NEW_SEGMENT_PREFIX = "new_segment_";
+ private static final String PARTITION_COLUMN = "memberId";
+ private static final String OFFLINE_TENENT_NAME = "DefaultTenant_OFFLINE";
+ private static final String NEW_SEGMENT_PREFIX = "new_segment_";
private final TableConfig.Builder _offlineBuilder = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE);
@@ -63,12 +61,8 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
ControllerConf config = getDefaultControllerConfiguration();
config.setTableMinReplicas(MIN_NUM_REPLICAS);
startController(config);
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_BROKER_INSTANCES, true);
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR,
- NUM_SERVER_INSTANCES, true);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_BROKER_INSTANCES, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_SERVER_INSTANCES, true);
_offlineBuilder.setTableName("testOfflineTable").setTimeColumnName("timeColumn").setTimeType("DAYS")
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("5");
@@ -78,22 +72,14 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
// Join 4 more servers as untagged
String[] instanceNames = {"Server_localhost_a", "Server_localhost_b", "Server_localhost_c", "Server_localhost_d"};
for (String instanceName : instanceNames) {
- ControllerRequestBuilderUtil
- .addFakeDataInstanceToAutoJoinHelixCluster(getHelixClusterName(), ZkStarter.DEFAULT_ZK_STR, instanceName,
- true);
- _helixAdmin.removeInstanceTag(getHelixClusterName(), instanceName, OFFLINE_TENENT_NAME);
+ addFakeServerInstanceToAutoJoinHelixCluster(instanceName, true);
+ _helixAdmin.removeInstanceTag(_helixClusterName, instanceName, OFFLINE_TENENT_NAME);
}
} catch (Exception e) {
e.printStackTrace();
}
}
- @AfterClass
- public void tearDown() {
- stopController();
- stopZk();
- }
-
@Test
public void testReplicaGroupRebalanceStrategy()
throws Exception {
@@ -134,8 +120,8 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
}
// Test replace
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_0", OFFLINE_TENENT_NAME);
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_a", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_0", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_a", OFFLINE_TENENT_NAME);
_helixResourceManager.rebalanceTable(TABLE_NAME, CommonConstants.Helix.TableType.OFFLINE, rebalanceUserConfig);
Assert.assertTrue(validateTableLevelReplicaGroupRebalance());
Assert.assertTrue(validateNumSegments(INITIAL_NUM_SEGMENTS));
@@ -148,8 +134,8 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
Assert.assertTrue(validateTableLevelReplicaGroupRebalance());
// Test replace again
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_a", OFFLINE_TENENT_NAME);
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_0", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_a", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_0", OFFLINE_TENENT_NAME);
_helixResourceManager.rebalanceTable(TABLE_NAME, CommonConstants.Helix.TableType.OFFLINE, rebalanceUserConfig);
Assert.assertTrue(validateTableLevelReplicaGroupRebalance());
Assert.assertTrue(validateNumSegments(INITIAL_NUM_SEGMENTS + 10));
@@ -161,10 +147,10 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
}
// Test adding servers to each replica group
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_a", OFFLINE_TENENT_NAME);
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_b", OFFLINE_TENENT_NAME);
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_c", OFFLINE_TENENT_NAME);
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_d", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_a", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_b", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_c", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_d", OFFLINE_TENENT_NAME);
int targetNumInstancePerPartition = 5;
int targetNumReplicaGroup = 2;
@@ -174,8 +160,8 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
Assert.assertTrue(validateNumSegments(INITIAL_NUM_SEGMENTS));
// Test removing servers to each replica group
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_a", OFFLINE_TENENT_NAME);
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_d", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_a", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_d", OFFLINE_TENENT_NAME);
targetNumInstancePerPartition = 4;
targetNumReplicaGroup = 2;
updateTableConfig(targetNumInstancePerPartition, targetNumReplicaGroup);
@@ -197,8 +183,8 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
}
// Test removing two more servers to each replica group with force run
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_b", OFFLINE_TENENT_NAME);
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_c", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_b", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_c", OFFLINE_TENENT_NAME);
targetNumInstancePerPartition = 3;
targetNumReplicaGroup = 2;
@@ -208,9 +194,9 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
Assert.assertTrue(validateNumSegments(INITIAL_NUM_SEGMENTS));
// Test adding a replica group
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_a", OFFLINE_TENENT_NAME);
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_b", OFFLINE_TENENT_NAME);
- _helixAdmin.addInstanceTag(getHelixClusterName(), "Server_localhost_c", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_a", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_b", OFFLINE_TENENT_NAME);
+ _helixAdmin.addInstanceTag(_helixClusterName, "Server_localhost_c", OFFLINE_TENENT_NAME);
targetNumInstancePerPartition = 3;
targetNumReplicaGroup = 3;
@@ -233,9 +219,9 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
}
// Test removing a replica group
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_0", OFFLINE_TENENT_NAME);
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_1", OFFLINE_TENENT_NAME);
- _helixAdmin.removeInstanceTag(getHelixClusterName(), "Server_localhost_2", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_0", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_1", OFFLINE_TENENT_NAME);
+ _helixAdmin.removeInstanceTag(_helixClusterName, "Server_localhost_2", OFFLINE_TENENT_NAME);
targetNumInstancePerPartition = 3;
targetNumReplicaGroup = 2;
@@ -263,7 +249,7 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
private boolean validateNumSegments(int numSegments) {
String tableNameWithType = TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME);
- IdealState idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), tableNameWithType);
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableNameWithType);
return idealState.getRecord().getMapFields().keySet().size() == numSegments;
}
@@ -274,7 +260,7 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
new ReplicaGroupPartitionAssignmentGenerator(_propertyStore);
ReplicaGroupPartitionAssignment replicaGroupMapping =
partitionAssignmentGenerator.getReplicaGroupPartitionAssignment(tableNameWithType);
- IdealState idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), tableNameWithType);
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableNameWithType);
Map<String, Map<String, String>> segmentAssignment = idealState.getRecord().getMapFields();
Map<Integer, Set<String>> segmentsPerPartition = new HashMap<>();
segmentsPerPartition.put(0, segmentAssignment.keySet());
@@ -309,7 +295,7 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
private boolean allSegmentsPushedToIdealState(String tableName, int segmentNum) {
IdealState idealState =
- _helixAdmin.getResourceIdealState(getHelixClusterName(), TableNameBuilder.OFFLINE.tableNameWithType(tableName));
+ _helixAdmin.getResourceIdealState(_helixClusterName, TableNameBuilder.OFFLINE.tableNameWithType(tableName));
return idealState != null && idealState.getPartitionSet() != null
&& idealState.getPartitionSet().size() == segmentNum;
}
@@ -324,4 +310,11 @@ public class ReplicaGroupRebalanceStrategyTest extends ControllerTest {
_helixResourceManager
.setExistingTableConfig(tableConfig, tableNameWithType, CommonConstants.Helix.TableType.OFFLINE);
}
+
+ @AfterClass
+ public void tearDown() {
+ stopFakeInstances();
+ stopController();
+ stopZk();
+ }
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/sharding/SegmentAssignmentStrategyTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/sharding/SegmentAssignmentStrategyTest.java
index 836824c..e12f5e8 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/sharding/SegmentAssignmentStrategyTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/sharding/SegmentAssignmentStrategyTest.java
@@ -22,8 +22,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.Set;
-import org.I0Itec.zkclient.ZkClient;
-import org.apache.helix.HelixManager;
import org.apache.helix.model.IdealState;
import org.apache.pinot.common.config.ColumnPartitionConfig;
import org.apache.pinot.common.config.IndexingConfig;
@@ -35,71 +33,39 @@ import org.apache.pinot.common.data.Schema;
import org.apache.pinot.common.partition.ReplicaGroupPartitionAssignment;
import org.apache.pinot.common.partition.ReplicaGroupPartitionAssignmentGenerator;
import org.apache.pinot.common.utils.CommonConstants;
-import org.apache.pinot.common.utils.ZkStarter;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.controller.utils.ReplicaGroupTestUtils;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
-import org.apache.pinot.core.realtime.stream.StreamConfig;
-import org.apache.pinot.core.realtime.stream.StreamConfigProperties;
import org.testng.Assert;
-import org.testng.annotations.AfterTest;
-import org.testng.annotations.BeforeTest;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
public class SegmentAssignmentStrategyTest extends ControllerTest {
- private final static String ZK_SERVER = ZkStarter.DEFAULT_ZK_STR;
- private final static String TABLE_NAME_BALANCED = "testResourceBalanced";
- private final static String TABLE_NAME_RANDOM = "testResourceRandom";
- private final static String TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT = "testReplicaGroupPartitionAssignment";
- private final static String TABLE_NAME_TABLE_LEVEL_REPLICA_GROUP = "testTableLevelReplicaGroup";
- private final static String TABLE_NAME_PARTITION_LEVEL_REPLICA_GROUP = "testPartitionLevelReplicaGroup";
-
- private static final Random random = new Random();
- private final static String PARTITION_COLUMN = "memberId";
- private final static int NUM_REPLICA = 2;
- private ZkClient _zkClient;
- private final int _numServerInstance = 10;
- private final int _numBrokerInstance = 1;
- private ZkStarter.ZookeeperInstance _zookeeperInstance;
+ private static final String TABLE_NAME_BALANCED = "testResourceBalanced";
+ private static final String TABLE_NAME_RANDOM = "testResourceRandom";
+ private static final String TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT = "testReplicaGroupPartitionAssignment";
+ private static final String TABLE_NAME_TABLE_LEVEL_REPLICA_GROUP = "testTableLevelReplicaGroup";
+ private static final String TABLE_NAME_PARTITION_LEVEL_REPLICA_GROUP = "testPartitionLevelReplicaGroup";
+ private static final String PARTITION_COLUMN = "memberId";
+ private static final int NUM_REPLICAS = 2;
+ private static final int NUM_SERVERS = 10;
+ private static final int NUM_BROKERS = 1;
+ private static final Random RANDOM = new Random();
+
private ReplicaGroupPartitionAssignmentGenerator _partitionAssignmentGenerator;
- @BeforeTest
- public void setup()
+ @BeforeClass
+ public void setUp()
throws Exception {
- _zookeeperInstance = ZkStarter.startLocalZkServer();
- _zkClient = new ZkClient(ZK_SERVER);
- final String zkPath = "/" + getHelixClusterName();
- if (_zkClient.exists(zkPath)) {
- _zkClient.deleteRecursive(zkPath);
- }
+ startZk();
startController();
- HelixManager helixZkManager = _helixResourceManager.getHelixZkManager();
-
- _partitionAssignmentGenerator =
- new ReplicaGroupPartitionAssignmentGenerator(helixZkManager.getHelixPropertyStore());
-
- ControllerRequestBuilderUtil
- .addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZK_SERVER, _numServerInstance, true);
- ControllerRequestBuilderUtil
- .addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZK_SERVER, _numBrokerInstance, true);
- Thread.sleep(100);
- Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_OFFLINE").size(),
- _numServerInstance);
- Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_REALTIME").size(),
- _numServerInstance);
-
- Assert.assertEquals(_helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), "DefaultTenant_BROKER").size(),
- _numBrokerInstance);
- }
+ addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_BROKERS, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(NUM_SERVERS, true);
- @AfterTest
- public void tearDown() {
- _helixResourceManager.stop();
- _zkClient.close();
- ZkStarter.stopLocalZkServer(_zookeeperInstance);
+ _partitionAssignmentGenerator = new ReplicaGroupPartitionAssignmentGenerator(_propertyStore);
}
@Test
@@ -108,7 +74,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
// Adding table
TableConfig tableConfig =
new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(TABLE_NAME_RANDOM)
- .setSegmentAssignmentStrategy("RandomAssignmentStrategy").setNumReplicas(NUM_REPLICA).build();
+ .setSegmentAssignmentStrategy("RandomAssignmentStrategy").setNumReplicas(NUM_REPLICAS).build();
_helixResourceManager.addTable(tableConfig);
// Wait for the table addition
@@ -125,17 +91,16 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
while (!allSegmentsPushedToIdealState(TABLE_NAME_RANDOM, i + 1)) {
Thread.sleep(100);
}
- final Set<String> taggedInstances =
- _helixResourceManager.getAllInstancesForServerTenant("DefaultTenant_OFFLINE");
+ final Set<String> taggedInstances = _helixResourceManager.getAllInstancesForServerTenant("DefaultTenant_OFFLINE");
final Map<String, Integer> instanceToNumSegmentsMap = new HashMap<>();
for (final String instance : taggedInstances) {
instanceToNumSegmentsMap.put(instance, 0);
}
IdealState idealState = _helixAdmin
- .getResourceIdealState(getHelixClusterName(), TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_RANDOM));
+ .getResourceIdealState(_helixClusterName, TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_RANDOM));
Assert.assertEquals(idealState.getPartitionSet().size(), i + 1);
for (final String segmentId : idealState.getPartitionSet()) {
- Assert.assertEquals(idealState.getInstanceStateMap(segmentId).size(), NUM_REPLICA);
+ Assert.assertEquals(idealState.getInstanceStateMap(segmentId).size(), NUM_REPLICAS);
}
}
}
@@ -163,23 +128,22 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
Thread.sleep(100);
}
- final Set<String> taggedInstances =
- _helixResourceManager.getAllInstancesForServerTenant("DefaultTenant_OFFLINE");
+ final Set<String> taggedInstances = _helixResourceManager.getAllInstancesForServerTenant("DefaultTenant_OFFLINE");
final Map<String, Integer> instance2NumSegmentsMap = new HashMap<>();
for (final String instance : taggedInstances) {
instance2NumSegmentsMap.put(instance, 0);
}
final IdealState idealState = _helixAdmin
- .getResourceIdealState(getHelixClusterName(), TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_BALANCED));
+ .getResourceIdealState(_helixClusterName, TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_BALANCED));
for (final String segmentId : idealState.getPartitionSet()) {
for (final String instance : idealState.getInstanceStateMap(segmentId).keySet()) {
instance2NumSegmentsMap.put(instance, instance2NumSegmentsMap.get(instance) + 1);
}
}
final int totalSegments = (numSegments) * numReplicas;
- final int minNumSegmentsPerInstance = totalSegments / _numServerInstance;
+ final int minNumSegmentsPerInstance = totalSegments / NUM_SERVERS;
int maxNumSegmentsPerInstance = minNumSegmentsPerInstance;
- if ((minNumSegmentsPerInstance * _numServerInstance) < totalSegments) {
+ if ((minNumSegmentsPerInstance * NUM_SERVERS) < totalSegments) {
maxNumSegmentsPerInstance = maxNumSegmentsPerInstance + 1;
}
for (final String instance : instance2NumSegmentsMap.keySet()) {
@@ -188,7 +152,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
Assert.assertTrue(instance2NumSegmentsMap.get(instance) <= maxNumSegmentsPerInstance,
"expected <=" + maxNumSegmentsPerInstance + " actual:" + instance2NumSegmentsMap.get(instance));
}
- _helixAdmin.dropResource(getHelixClusterName(), TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_BALANCED));
+ _helixAdmin.dropResource(_helixClusterName, TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_BALANCED));
}
@Test
@@ -200,7 +164,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
// Adding a table without replica group
TableConfig tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE)
.setTableName(TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT)
- .setSegmentAssignmentStrategy("RandomAssignmentStrategy").setNumReplicas(NUM_REPLICA).build();
+ .setSegmentAssignmentStrategy("RandomAssignmentStrategy").setNumReplicas(NUM_REPLICAS).build();
_helixResourceManager.addTable(tableConfig);
// Check that partition assignment does not exist
@@ -216,7 +180,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
replicaGroupStrategyConfig.setMirrorAssignmentAcrossReplicaGroups(true);
TableConfig replicaGroupTableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE)
- .setTableName(TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT).setNumReplicas(NUM_REPLICA)
+ .setTableName(TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT).setNumReplicas(NUM_REPLICAS)
.setSegmentAssignmentStrategy("ReplicaGroupSegmentAssignmentStrategy").build();
replicaGroupTableConfig.getValidationConfig().setReplicaGroupStrategyConfig(replicaGroupStrategyConfig);
@@ -257,13 +221,10 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
Map<String, String> streamConfigMap = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
// Adding a table without replica group
- TableConfig tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.REALTIME).setTableName(
- TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT)
- .setSegmentAssignmentStrategy("RandomAssignmentStrategy")
- .setNumReplicas(NUM_REPLICA)
- .setStreamConfigs(streamConfigMap)
- .setLLC(true)
- .build();
+ TableConfig tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.REALTIME)
+ .setTableName(TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT)
+ .setSegmentAssignmentStrategy("RandomAssignmentStrategy").setNumReplicas(NUM_REPLICAS)
+ .setStreamConfigs(streamConfigMap).setLLC(true).build();
try {
_helixResourceManager.addTable(tableConfig);
} catch (Exception e) {
@@ -282,20 +243,17 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
replicaGroupStrategyConfig.setNumInstancesPerPartition(numInstancesPerPartition);
replicaGroupStrategyConfig.setMirrorAssignmentAcrossReplicaGroups(true);
- TableConfig replicaGroupTableConfig =
- new TableConfig.Builder(CommonConstants.Helix.TableType.REALTIME).setTableName(
- TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT)
- .setNumReplicas(NUM_REPLICA)
- .setSegmentAssignmentStrategy("ReplicaGroupSegmentAssignmentStrategy")
- .setStreamConfigs(streamConfigMap)
- .setLLC(true)
- .build();
+ TableConfig replicaGroupTableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.REALTIME)
+ .setTableName(TABLE_NAME_REPLICA_GROUP_PARTITION_ASSIGNMENT).setNumReplicas(NUM_REPLICAS)
+ .setSegmentAssignmentStrategy("ReplicaGroupSegmentAssignmentStrategy").setStreamConfigs(streamConfigMap)
+ .setLLC(true).build();
replicaGroupTableConfig.getValidationConfig().setReplicaGroupStrategyConfig(replicaGroupStrategyConfig);
// Check that the replica group partition assignment is created
try {
- _helixResourceManager.setExistingTableConfig(replicaGroupTableConfig, tableNameWithType, CommonConstants.Helix.TableType.REALTIME);
+ _helixResourceManager
+ .setExistingTableConfig(replicaGroupTableConfig, tableNameWithType, CommonConstants.Helix.TableType.REALTIME);
} catch (Exception e) {
// ignore
}
@@ -333,7 +291,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
// Create table config
TableConfig tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE)
- .setTableName(TABLE_NAME_TABLE_LEVEL_REPLICA_GROUP).setNumReplicas(NUM_REPLICA)
+ .setTableName(TABLE_NAME_TABLE_LEVEL_REPLICA_GROUP).setNumReplicas(NUM_REPLICAS)
.setSegmentAssignmentStrategy("ReplicaGroupSegmentAssignmentStrategy").build();
tableConfig.getValidationConfig().setReplicaGroupStrategyConfig(replicaGroupStrategyConfig);
@@ -361,7 +319,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
ReplicaGroupPartitionAssignment partitionAssignment =
_partitionAssignmentGenerator.getReplicaGroupPartitionAssignment(offlineTableName);
- IdealState idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(),
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName,
TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_TABLE_LEVEL_REPLICA_GROUP));
// Validate the segment assignment
@@ -373,9 +331,9 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
@Test
public void testPartitionLevelReplicaGroupSegmentAssignmentStrategy()
throws Exception {
- int totalPartitionNumber = random.nextInt(8) + 2;
- int numInstancesPerPartition = random.nextInt(5) + 1;
- int numSegments = random.nextInt(10) + 10;
+ int totalPartitionNumber = RANDOM.nextInt(8) + 2;
+ int numInstancesPerPartition = RANDOM.nextInt(5) + 1;
+ int numSegments = RANDOM.nextInt(10) + 10;
// Create the configuration for segment assignment strategy.
ReplicaGroupStrategyConfig replicaGroupStrategyConfig = new ReplicaGroupStrategyConfig();
@@ -392,7 +350,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
// Create table config
TableConfig tableConfig = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE)
- .setTableName(TABLE_NAME_PARTITION_LEVEL_REPLICA_GROUP).setNumReplicas(NUM_REPLICA)
+ .setTableName(TABLE_NAME_PARTITION_LEVEL_REPLICA_GROUP).setNumReplicas(NUM_REPLICAS)
.setSegmentAssignmentStrategy("ReplicaGroupSegmentAssignmentStrategy").build();
tableConfig.getValidationConfig().setReplicaGroupStrategyConfig(replicaGroupStrategyConfig);
@@ -421,7 +379,7 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
ReplicaGroupPartitionAssignment partitionAssignment =
_partitionAssignmentGenerator.getReplicaGroupPartitionAssignment(offlineTable);
- IdealState idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(),
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName,
TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME_PARTITION_LEVEL_REPLICA_GROUP));
// Validate the segment assignment
@@ -432,8 +390,15 @@ public class SegmentAssignmentStrategyTest extends ControllerTest {
private boolean allSegmentsPushedToIdealState(String tableName, int segmentNum) {
IdealState idealState =
- _helixAdmin.getResourceIdealState(getHelixClusterName(), TableNameBuilder.OFFLINE.tableNameWithType(tableName));
+ _helixAdmin.getResourceIdealState(_helixClusterName, TableNameBuilder.OFFLINE.tableNameWithType(tableName));
return idealState != null && idealState.getPartitionSet() != null
&& idealState.getPartitionSet().size() == segmentNum;
}
+
+ @AfterClass
+ public void tearDown() {
+ stopFakeInstances();
+ stopController();
+ stopZk();
+ }
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java
index 55c13a5..bf822f4 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java
@@ -21,8 +21,6 @@ package org.apache.pinot.controller.validation;
import java.util.ArrayList;
import java.util.List;
import org.apache.helix.HelixAdmin;
-import org.apache.helix.HelixManager;
-import org.apache.helix.manager.zk.ZkClient;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.config.TableConfig;
@@ -33,9 +31,7 @@ import org.apache.pinot.common.segment.SegmentMetadata;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.HLCSegmentName;
import org.apache.pinot.common.utils.LLCSegmentName;
-import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.common.utils.helix.HelixHelper;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
import org.joda.time.DateTime;
@@ -54,33 +50,23 @@ import static org.testng.Assert.assertEquals;
* Tests for the ValidationManagers.
*/
public class ValidationManagerTest extends ControllerTest {
- private static final String ZK_STR = ZkStarter.DEFAULT_ZK_STR;
private static final String TEST_TABLE_NAME = "testTable";
private static final String TEST_TABLE_TWO = "testTable2";
private static final String TEST_SEGMENT_NAME = "testSegment";
- private ZkClient _zkClient;
-
- private ZkStarter.ZookeeperInstance _zookeeperInstance;
private TableConfig _offlineTableConfig;
- private HelixManager _helixManager;
@BeforeClass
public void setUp()
throws Exception {
- _zookeeperInstance = ZkStarter.startLocalZkServer();
- _zkClient = new ZkClient(ZK_STR);
- Thread.sleep(1000);
-
+ startZk();
startController();
-
- ControllerRequestBuilderUtil.addFakeDataInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZK_STR, 2, true);
- ControllerRequestBuilderUtil.addFakeBrokerInstancesToAutoJoinHelixCluster(getHelixClusterName(), ZK_STR, 2, true);
+ addFakeBrokerInstancesToAutoJoinHelixCluster(2, true);
+ addFakeServerInstancesToAutoJoinHelixCluster(2, true);
_offlineTableConfig =
new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(TEST_TABLE_NAME).setNumReplicas(2)
.build();
- _helixManager = _helixResourceManager.getHelixZkManager();
_helixResourceManager.addTable(_offlineTableConfig);
}
@@ -91,7 +77,7 @@ public class ValidationManagerTest extends ControllerTest {
String partitionName = _offlineTableConfig.getTableName();
HelixAdmin helixAdmin = _helixManager.getClusterManagmentTool();
- IdealState idealState = HelixHelper.getBrokerIdealStates(helixAdmin, getHelixClusterName());
+ IdealState idealState = HelixHelper.getBrokerIdealStates(helixAdmin, _helixClusterName);
// Ensure that the broker resource is not rebuilt.
Assert.assertTrue(idealState.getInstanceSet(partitionName)
.equals(_helixResourceManager.getAllInstancesForBrokerTenant(TagNameUtils.DEFAULT_TENANT_NAME)));
@@ -109,15 +95,15 @@ public class ValidationManagerTest extends ControllerTest {
instanceConfig.setInstanceEnabled(true);
instanceConfig.setHostName("Broker_localhost");
instanceConfig.setPort("2");
- helixAdmin.addInstance(getHelixClusterName(), instanceConfig);
- helixAdmin.addInstanceTag(getHelixClusterName(), instanceConfig.getInstanceName(),
+ helixAdmin.addInstance(_helixClusterName, instanceConfig);
+ helixAdmin.addInstanceTag(_helixClusterName, instanceConfig.getInstanceName(),
TagNameUtils.getBrokerTagForTenant(TagNameUtils.DEFAULT_TENANT_NAME));
- idealState = HelixHelper.getBrokerIdealStates(helixAdmin, getHelixClusterName());
+ idealState = HelixHelper.getBrokerIdealStates(helixAdmin, _helixClusterName);
// Assert that the two don't equal before the call to rebuild the broker resource.
Assert.assertTrue(!idealState.getInstanceSet(partitionNameTwo)
.equals(_helixResourceManager.getAllInstancesForBrokerTenant(TagNameUtils.DEFAULT_TENANT_NAME)));
_helixResourceManager.rebuildBrokerResourceFromHelixTags(partitionNameTwo);
- idealState = HelixHelper.getBrokerIdealStates(helixAdmin, getHelixClusterName());
+ idealState = HelixHelper.getBrokerIdealStates(helixAdmin, _helixClusterName);
// Assert that the two do equal after being rebuilt.
Assert.assertTrue(idealState.getInstanceSet(partitionNameTwo)
.equals(_helixResourceManager.getAllInstancesForBrokerTenant(TagNameUtils.DEFAULT_TENANT_NAME)));
@@ -141,8 +127,7 @@ public class ValidationManagerTest extends ControllerTest {
Mockito.when(segmentMetadata.getCrc()).thenReturn(Long.toString(System.nanoTime()));
_helixResourceManager.refreshSegment(TEST_TABLE_NAME, segmentMetadata, offlineSegmentZKMetadata);
- offlineSegmentZKMetadata =
- _helixResourceManager.getOfflineSegmentZKMetadata(TEST_TABLE_NAME, TEST_SEGMENT_NAME);
+ offlineSegmentZKMetadata = _helixResourceManager.getOfflineSegmentZKMetadata(TEST_TABLE_NAME, TEST_SEGMENT_NAME);
// Check that the segment still has the same push time
assertEquals(offlineSegmentZKMetadata.getPushTime(), pushTime);
// Check that the refresh time is in the last 30 seconds
@@ -186,13 +171,6 @@ public class ValidationManagerTest extends ControllerTest {
15);
}
- @AfterClass
- public void shutDown() {
- _helixResourceManager.stop();
- _zkClient.close();
- ZkStarter.stopLocalZkServer(_zookeeperInstance);
- }
-
@Test
public void testComputeNumMissingSegments() {
Interval jan1st = new Interval(new DateTime(2015, 1, 1, 0, 0, 0), new DateTime(2015, 1, 1, 23, 59, 59));
@@ -226,4 +204,11 @@ public class ValidationManagerTest extends ControllerTest {
jan1st2nd4th5th.add(jan5th);
assertEquals(OfflineSegmentIntervalChecker.computeNumMissingSegments(jan1st2nd4th5th, Duration.standardDays(1)), 1);
}
+
+ @AfterClass
+ public void tearDown() {
+ stopFakeInstances();
+ stopController();
+ stopZk();
+ }
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BalanceNumSegmentAssignmentStrategyIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BalanceNumSegmentAssignmentStrategyIntegrationTest.java
index b8ac99a..992784e 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BalanceNumSegmentAssignmentStrategyIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BalanceNumSegmentAssignmentStrategyIntegrationTest.java
@@ -84,7 +84,7 @@ public class BalanceNumSegmentAssignmentStrategyIntegrationTest extends UploadRe
private Map<String, Integer> getSegmentsPerInstance(String tableName) {
Map<String, Integer> segmentsPerInstance = new HashMap<String, Integer>();
- IdealState idealState = _helixAdmin.getResourceIdealState(getHelixClusterName(), tableName + "_OFFLINE");
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableName + "_OFFLINE");
for (String partitionName : idealState.getPartitionSet()) {
for (String instanceName : idealState.getInstanceSet(partitionName)) {
if (!segmentsPerInstance.containsKey(instanceName)) {
@@ -100,12 +100,12 @@ public class BalanceNumSegmentAssignmentStrategyIntegrationTest extends UploadRe
@Test(dataProvider = "tableNameProvider")
public void testNoAssignmentToDisabledInstances(String tableName, SegmentVersion version)
throws Exception {
- List<String> instances = _helixAdmin.getInstancesInClusterWithTag(getHelixClusterName(), serverTenant);
+ List<String> instances = _helixAdmin.getInstancesInClusterWithTag(_helixClusterName, serverTenant);
List<String> disabledInstances = new ArrayList<>();
// disable 6 instances
assertEquals(instances.size(), 9);
for (int i = 0; i < 6; i++) {
- _helixAdmin.enableInstance(getHelixClusterName(), instances.get(i), false);
+ _helixAdmin.enableInstance(_helixClusterName, instances.get(i), false);
}
// Thread.sleep(100000);
@@ -122,7 +122,7 @@ public class BalanceNumSegmentAssignmentStrategyIntegrationTest extends UploadRe
// re-enable instances since these tests are usually "setup once" and run multiple tests type
for (int i = 0; i < 6; i++) {
- _helixAdmin.enableInstance(getHelixClusterName(), instances.get(i), true);
+ _helixAdmin.enableInstance(_helixClusterName, instances.get(i), true);
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index 70f3d7f..c321a8c 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -84,7 +84,6 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
private Connection _h2Connection;
private QueryGenerator _queryGenerator;
-
/**
* The following getters can be overridden to change default settings.
*/
@@ -216,7 +215,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
@Nonnull
protected org.apache.pinot.client.Connection getPinotConnection() {
if (_pinotConnection == null) {
- _pinotConnection = ConnectionFactory.fromZookeeper(ZkStarter.DEFAULT_ZK_STR + "/" + getHelixClusterName());
+ _pinotConnection = ConnectionFactory.fromZookeeper(ZkStarter.DEFAULT_ZK_STR + "/" + _helixClusterName);
}
return _pinotConnection;
}
@@ -324,9 +323,8 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
@Override
public void run() {
try {
- ClusterIntegrationTestUtils
- .pushAvroIntoKafka(avroFiles, KafkaStarterUtils.DEFAULT_KAFKA_BROKER, kafkaTopic,
- getMaxNumKafkaMessagesPerBatch(), getKafkaMessageHeader(), getPartitionColumn());
+ ClusterIntegrationTestUtils.pushAvroIntoKafka(avroFiles, KafkaStarterUtils.DEFAULT_KAFKA_BROKER, kafkaTopic,
+ getMaxNumKafkaMessagesPerBatch(), getKafkaMessageHeader(), getPartitionColumn());
} catch (Exception e) {
// Ignored
}
@@ -336,8 +334,8 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
protected void startKafka() {
- _kafkaStarters =
- KafkaStarterUtils.startServers(getNumKafkaBrokers(), KafkaStarterUtils.DEFAULT_KAFKA_PORT, KafkaStarterUtils.DEFAULT_ZK_STR,
+ _kafkaStarters = KafkaStarterUtils
+ .startServers(getNumKafkaBrokers(), KafkaStarterUtils.DEFAULT_KAFKA_PORT, KafkaStarterUtils.DEFAULT_ZK_STR,
KafkaStarterUtils.getDefaultKafkaConfiguration());
_kafkaStarters.get(0)
.createTopic(getKafkaTopic(), KafkaStarterUtils.getTopicCreationProps(getNumKafkaPartitions()));
@@ -423,14 +421,14 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
addRealtimeTable(getTableName(), useLlc(), KafkaStarterUtils.DEFAULT_KAFKA_BROKER, KafkaStarterUtils.DEFAULT_ZK_STR,
getKafkaTopic(), getRealtimeSegmentFlushSize(), avroFile, timeColumnName, timeType, schemaName,
- getBrokerTenant(), getServerTenant(), getLoadMode(), getSortedColumn(),
- getInvertedIndexColumns(), getBloomFilterIndexColumns(), getRawIndexColumns(), getTaskConfig(),
- getStreamConsumerFactoryClassName());
+ getBrokerTenant(), getServerTenant(), getLoadMode(), getSortedColumn(), getInvertedIndexColumns(),
+ getBloomFilterIndexColumns(), getRawIndexColumns(), getTaskConfig(), getStreamConsumerFactoryClassName());
completeTableConfiguration();
}
- protected void completeTableConfiguration() throws IOException {
+ protected void completeTableConfiguration()
+ throws IOException {
if (isUsingNewConfigFormat()) {
CombinedConfig combinedConfig = new CombinedConfig(_offlineTableConfig, _realtimeTableConfig, _schema);
sendPostRequest(_controllerRequestURLBuilder.forNewTableCreate(), Serializer.serializeToString(combinedConfig));
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java
index 857a16d..e6fb44b 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java
@@ -141,7 +141,6 @@ public abstract class BaseClusterIntegrationTestSet extends BaseClusterIntegrati
+ "HAVING SUM(ArrDelay) <> 6325.973 AND AVG(CAST(CRSDepTime AS DOUBLE)) <= 1569.8755 OR SUM(TaxiIn) = 1003.87274"));
}
-
/**
* Test hardcoded queries.
* <p>NOTE:
@@ -361,14 +360,14 @@ public abstract class BaseClusterIntegrationTestSet extends BaseClusterIntegrati
*/
public void testInstanceShutdown()
throws Exception {
- List<String> instances = _helixAdmin.getInstancesInCluster(_clusterName);
+ List<String> instances = _helixAdmin.getInstancesInCluster(_helixClusterName);
Assert.assertFalse(instances.isEmpty(), "List of instances should not be empty");
// Mark all instances in the cluster as shutting down
for (String instance : instances) {
- InstanceConfig instanceConfig = _helixAdmin.getInstanceConfig(_clusterName, instance);
+ InstanceConfig instanceConfig = _helixAdmin.getInstanceConfig(_helixClusterName, instance);
instanceConfig.getRecord().setBooleanField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS, true);
- _helixAdmin.setInstanceConfig(_clusterName, instance, instanceConfig);
+ _helixAdmin.setInstanceConfig(_helixClusterName, instance, instanceConfig);
}
// Check that the routing table is empty
@@ -376,9 +375,9 @@ public abstract class BaseClusterIntegrationTestSet extends BaseClusterIntegrati
// Mark all instances as not shutting down
for (String instance : instances) {
- InstanceConfig instanceConfig = _helixAdmin.getInstanceConfig(_clusterName, instance);
+ InstanceConfig instanceConfig = _helixAdmin.getInstanceConfig(_helixClusterName, instance);
instanceConfig.getRecord().setBooleanField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS, false);
- _helixAdmin.setInstanceConfig(_clusterName, instance, instanceConfig);
+ _helixAdmin.setInstanceConfig(_helixClusterName, instance, instanceConfig);
}
// Check that the routing table is not empty
@@ -394,16 +393,16 @@ public abstract class BaseClusterIntegrationTestSet extends BaseClusterIntegrati
checkForInstanceInRoutingTable(true, instanceName);
// Mark the server instance as shutting down
- InstanceConfig instanceConfig = _helixAdmin.getInstanceConfig(_clusterName, instanceName);
+ InstanceConfig instanceConfig = _helixAdmin.getInstanceConfig(_helixClusterName, instanceName);
instanceConfig.getRecord().setBooleanField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS, true);
- _helixAdmin.setInstanceConfig(_clusterName, instanceName, instanceConfig);
+ _helixAdmin.setInstanceConfig(_helixClusterName, instanceName, instanceConfig);
// Check that it is not in the routing table
checkForInstanceInRoutingTable(false, instanceName);
// Re-enable the server instance
instanceConfig.getRecord().setBooleanField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS, false);
- _helixAdmin.setInstanceConfig(_clusterName, instanceName, instanceConfig);
+ _helixAdmin.setInstanceConfig(_helixClusterName, instanceName, instanceConfig);
// Check that it is in the routing table
checkForInstanceInRoutingTable(true, instanceName);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
index 8a6aa19..e90de36 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
@@ -60,7 +60,6 @@ import org.apache.pinot.common.utils.CommonConstants.Server;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
import org.apache.pinot.common.utils.JsonUtils;
import org.apache.pinot.common.utils.ZkStarter;
-import org.apache.pinot.controller.helix.ControllerRequestBuilderUtil;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.core.data.GenericRow;
import org.apache.pinot.core.indexsegment.generator.SegmentVersion;
@@ -88,7 +87,6 @@ public abstract class ClusterTest extends ControllerTest {
private static final Random RANDOM = new Random();
private static final int DEFAULT_BROKER_PORT = 18099;
- protected final String _clusterName = getHelixClusterName();
protected String _brokerBaseApiUrl;
private List<HelixBrokerStarter> _brokerStarters = new ArrayList<>();
@@ -129,7 +127,7 @@ public abstract class ClusterTest extends ControllerTest {
brokerConf.setProperty(Broker.CONFIG_OF_REQUEST_HANDLER_TYPE, Broker.SINGLE_CONNECTION_REQUEST_HANDLER_TYPE);
}
overrideBrokerConf(brokerConf);
- HelixBrokerStarter brokerStarter = new HelixBrokerStarter(brokerConf, _clusterName, zkStr);
+ HelixBrokerStarter brokerStarter = new HelixBrokerStarter(brokerConf, _helixClusterName, zkStr, LOCAL_HOST);
brokerStarter.start();
_brokerStarters.add(brokerStarter);
}
@@ -171,7 +169,7 @@ public abstract class ClusterTest extends ControllerTest {
.setProperty(Server.CONFIG_OF_INSTANCE_SEGMENT_TAR_DIR, Server.DEFAULT_INSTANCE_SEGMENT_TAR_DIR + "-" + i);
configuration.setProperty(Server.CONFIG_OF_ADMIN_API_PORT, baseAdminApiPort - i);
configuration.setProperty(Server.CONFIG_OF_NETTY_PORT, baseNettyPort + i);
- _serverStarters.add(new HelixServerStarter(_clusterName, zkStr, configuration));
+ _serverStarters.add(new HelixServerStarter(_helixClusterName, zkStr, configuration));
}
} catch (Exception e) {
throw new RuntimeException(e);
@@ -191,7 +189,7 @@ public abstract class ClusterTest extends ControllerTest {
config.setProperty(Helix.Instance.INSTANCE_ID_KEY,
Minion.INSTANCE_PREFIX + "minion" + i + "_" + (Minion.DEFAULT_HELIX_PORT + i));
config.setProperty(Helix.Instance.DATA_DIR_KEY, Minion.DEFAULT_INSTANCE_DATA_DIR + "-" + i);
- MinionStarter minionStarter = new MinionStarter(ZkStarter.DEFAULT_ZK_STR, _clusterName, config);
+ MinionStarter minionStarter = new MinionStarter(ZkStarter.DEFAULT_ZK_STR, _helixClusterName, config);
// Register task executor factories
if (taskExecutorFactoryRegistry != null) {
@@ -496,20 +494,6 @@ public abstract class ClusterTest extends ControllerTest {
bloomFilterColumns, noDictionaryColumns, taskConfig, streamConsumerFactoryName);
}
- protected void createBrokerTenant(String tenantName, int brokerCount)
- throws Exception {
- String request = ControllerRequestBuilderUtil.buildBrokerTenantCreateRequestJSON(tenantName, brokerCount);
- sendPostRequest(_controllerRequestURLBuilder.forBrokerTenantCreate(), request);
- }
-
- protected void createServerTenant(String tenantName, int offlineServerCount, int realtimeServerCount)
- throws Exception {
- String request = ControllerRequestBuilderUtil
- .buildServerTenantCreateRequestJSON(tenantName, offlineServerCount + realtimeServerCount, offlineServerCount,
- realtimeServerCount);
- sendPostRequest(_controllerRequestURLBuilder.forServerTenantCreate(), request);
- }
-
protected JsonNode getDebugInfo(final String uri)
throws Exception {
return JsonUtils.stringToJsonNode(sendGetRequest(_brokerBaseApiUrl + "/" + uri));
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 5281911..3434e93 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
@@ -90,15 +90,19 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
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;
}
@@ -119,7 +123,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
* @throws Exception
*/
@BeforeClass
- public void setUp() throws Exception {
+ public void setUp()
+ throws Exception {
TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
startZk();
@@ -159,7 +164,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
/**
* Setup offline table, but no segments
*/
- private void setupOfflineTable(String table) throws Exception {
+ private void setupOfflineTable(String table)
+ throws Exception {
_realtimeTableConfig = null;
addOfflineTable(table, null, null, TENANT_NAME, TENANT_NAME, null, SegmentVersion.v1, null, null, null, null, null);
completeTableConfiguration();
@@ -168,7 +174,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
/**
* Setup offline table, with segments from avro
*/
- private void setupOfflineTableAndSegments(String tableName, List<File> avroFiles) throws Exception {
+ private void setupOfflineTableAndSegments(String tableName, List<File> avroFiles)
+ throws Exception {
TestUtils.ensureDirectoriesExistAndEmpty(_segmentDir, _tarDir);
setTableName(tableName);
_realtimeTableConfig = null;
@@ -184,12 +191,13 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
Assert.assertNotNull(outgoingTimeUnit);
String timeType = outgoingTimeUnit.toString();
- addOfflineTable(tableName, timeColumnName, timeType, TENANT_NAME, TENANT_NAME, null, SegmentVersion.v1, null, null, null, null, null);
+ addOfflineTable(tableName, timeColumnName, timeType, TENANT_NAME, TENANT_NAME, null, SegmentVersion.v1, null, null,
+ null, null, null);
completeTableConfiguration();
ExecutorService executor = Executors.newCachedThreadPool();
- ClusterIntegrationTestUtils.buildSegmentsFromAvro(avroFiles, 0, _segmentDir, _tarDir, tableName, false,
- null, null, null, executor);
+ ClusterIntegrationTestUtils
+ .buildSegmentsFromAvro(avroFiles, 0, _segmentDir, _tarDir, tableName, false, null, null, null, executor);
executor.shutdown();
executor.awaitTermination(10, TimeUnit.MINUTES);
uploadSegments(getTableName(), _tarDir);
@@ -199,7 +207,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
/**
* Setup realtime table for given tablename and topic
*/
- private void setupRealtimeTable(String table, String topic, File avroFile) throws Exception {
+ private void setupRealtimeTable(String table, String topic, File avroFile)
+ throws Exception {
_offlineTableConfig = null;
File schemaFile = getSchemaFile();
Schema schema = Schema.fromFile(schemaFile);
@@ -233,7 +242,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
* @throws Exception
*/
@BeforeGroups(groups = "segmentStatusChecker")
- public void beforeTestSegmentStatusCheckerTest(ITestContext context) throws Exception {
+ public void beforeTestSegmentStatusCheckerTest(ITestContext context)
+ throws Exception {
String emptyTable = "table1_OFFLINE";
String disabledOfflineTable = "table2_OFFLINE";
String basicOfflineTable = getDefaultOfflineTableName();
@@ -253,7 +263,7 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
// table with disabled ideal state
setupOfflineTable(disabledOfflineTable);
- _helixAdmin.enableResource(_clusterName, disabledOfflineTable, false);
+ _helixAdmin.enableResource(_helixClusterName, disabledOfflineTable, false);
// some segments offline
setupOfflineTableAndSegments(errorOfflineTable, _avroFiles);
@@ -282,7 +292,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
* Validate that we are seeing the expected numbers
*/
@Test(groups = "segmentStatusChecker")
- public void testSegmentStatusChecker(ITestContext context) throws Exception {
+ public void testSegmentStatusChecker(ITestContext context)
+ throws Exception {
String emptyTable = (String) context.getAttribute("emptyTable");
String disabledOfflineTable = (String) context.getAttribute("disabledOfflineTable");
String basicOfflineTable = (String) context.getAttribute("basicOfflineTable");
@@ -292,9 +303,9 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
ControllerMetrics controllerMetrics = _controllerStarter.getControllerMetrics();
- TestUtils.waitForCondition(input ->
- controllerMetrics.getValueOfGlobalGauge(ControllerGauge.PERIODIC_TASK_NUM_TABLES_PROCESSED,
- "SegmentStatusChecker") >= numTables, 240_000, "Timed out waiting for SegmentStatusChecker");
+ TestUtils.waitForCondition(input -> controllerMetrics
+ .getValueOfGlobalGauge(ControllerGauge.PERIODIC_TASK_NUM_TABLES_PROCESSED, "SegmentStatusChecker") >= numTables,
+ 240_000, "Timed out waiting for SegmentStatusChecker");
// empty table - table1_OFFLINE
// num replicas set from ideal state
@@ -344,7 +355,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
}
@AfterGroups(groups = "segmentStatusChecker")
- public void afterTestSegmentStatusChecker(ITestContext context) throws Exception {
+ public void afterTestSegmentStatusChecker(ITestContext context)
+ throws Exception {
String emptyTable = (String) context.getAttribute("emptyTable");
String disabledOfflineTable = (String) context.getAttribute("disabledOfflineTable");
String errorOfflineTable = (String) context.getAttribute("errorOfflineTable");
@@ -361,7 +373,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
* @throws Exception
*/
@BeforeGroups(groups = "realtimeSegmentRelocator", dependsOnGroups = "segmentStatusChecker")
- public void beforeRealtimeSegmentRelocatorTest(ITestContext context) throws Exception {
+ public void beforeRealtimeSegmentRelocatorTest(ITestContext context)
+ throws Exception {
String relocationTable = getDefaultRealtimeTableName();
context.setAttribute("relocationTable", relocationTable);
@@ -377,20 +390,22 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
}
@Test(groups = "realtimeSegmentRelocator", dependsOnGroups = "segmentStatusChecker")
- public void testRealtimeSegmentRelocator(ITestContext context) throws Exception {
+ public void testRealtimeSegmentRelocator(ITestContext context)
+ throws Exception {
String relocationTable = (String) context.getAttribute("relocationTable");
ControllerMetrics controllerMetrics = _controllerStarter.getControllerMetrics();
- long taskRunCount = controllerMetrics.getMeteredTableValue("RealtimeSegmentRelocator",
- ControllerMeter.CONTROLLER_PERIODIC_TASK_RUN).count();
+ long taskRunCount =
+ controllerMetrics.getMeteredTableValue("RealtimeSegmentRelocator", ControllerMeter.CONTROLLER_PERIODIC_TASK_RUN)
+ .count();
TestUtils.waitForCondition(input ->
controllerMetrics.getMeteredTableValue("RealtimeSegmentRelocator", ControllerMeter.CONTROLLER_PERIODIC_TASK_RUN)
.count() > taskRunCount, 60_000, "Timed out waiting for RealtimeSegmentRelocation to run");
- Assert.assertTrue(controllerMetrics.getValueOfGlobalGauge(ControllerGauge.PERIODIC_TASK_NUM_TABLES_PROCESSED,
- "RealtimeSegmentRelocator") > 0);
+ Assert.assertTrue(controllerMetrics
+ .getValueOfGlobalGauge(ControllerGauge.PERIODIC_TASK_NUM_TABLES_PROCESSED, "RealtimeSegmentRelocator") > 0);
// check servers for ONLINE segment and CONSUMING segments are disjoint sets
Set<String> consuming = new HashSet<>();
@@ -425,12 +440,13 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
// Check that the first table we added doesn't need to be rebuilt(case where ideal state brokers and brokers in broker resource are the same.
String table1 = (String) context.getAttribute("testTableOne");
String table2 = (String) context.getAttribute("testTableTwo");
- TableConfig tableConfigOne = new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(table1).build();
+ TableConfig tableConfigOne =
+ new TableConfig.Builder(CommonConstants.Helix.TableType.OFFLINE).setTableName(table1).build();
String partitionNameOne = tableConfigOne.getTableName();
// Ensure that the broker resource is not rebuilt.
TestUtils.waitForCondition(input -> {
- IdealState idealState = HelixHelper.getBrokerIdealStates(_helixAdmin, getHelixClusterName());
+ IdealState idealState = HelixHelper.getBrokerIdealStates(_helixAdmin, _helixClusterName);
return idealState.getInstanceSet(partitionNameOne)
.equals(_helixResourceManager.getAllInstancesForBrokerTenant(TENANT_NAME));
}, 60_000L, "Timeout when waiting for broker resource to be rebuilt");
@@ -448,15 +464,15 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
instanceConfig.setInstanceEnabled(true);
instanceConfig.setHostName("Broker_localhost");
instanceConfig.setPort("2");
- _helixAdmin.addInstance(getHelixClusterName(), instanceConfig);
- _helixAdmin.addInstanceTag(getHelixClusterName(), instanceConfig.getInstanceName(),
+ _helixAdmin.addInstance(_helixClusterName, instanceConfig);
+ _helixAdmin.addInstanceTag(_helixClusterName, instanceConfig.getInstanceName(),
TagNameUtils.getBrokerTagForTenant(TENANT_NAME));
// Count the number of times we check on ideal state change, which is made by rebuild broker resource method.
AtomicInteger count = new AtomicInteger();
TestUtils.waitForCondition(input -> {
count.getAndIncrement();
- IdealState idealState = HelixHelper.getBrokerIdealStates(_helixAdmin, getHelixClusterName());
+ IdealState idealState = HelixHelper.getBrokerIdealStates(_helixAdmin, _helixClusterName);
return idealState.getInstanceSet(partitionNameTwo)
.equals(_helixResourceManager.getAllInstancesForBrokerTenant(TENANT_NAME));
}, 60_000L, "Timeout when waiting for broker resource to be rebuilt");
@@ -465,11 +481,11 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
Assert.assertTrue(count.get() > 1);
// Drop the instance so that broker resource doesn't match the current one.
- _helixAdmin.dropInstance(getHelixClusterName(), instanceConfig);
+ _helixAdmin.dropInstance(_helixClusterName, instanceConfig);
count.set(0);
TestUtils.waitForCondition(input -> {
count.getAndIncrement();
- IdealState idealState = HelixHelper.getBrokerIdealStates(_helixAdmin, getHelixClusterName());
+ IdealState idealState = HelixHelper.getBrokerIdealStates(_helixAdmin, _helixClusterName);
return idealState.getInstanceSet(partitionNameTwo)
.equals(_helixResourceManager.getAllInstancesForBrokerTenant(TENANT_NAME));
}, 60_000L, "Timeout when waiting for broker resource to be rebuilt");
@@ -517,27 +533,29 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
* @throws Exception
*/
- @Test(groups="realtimeSegmentValidationManager", dependsOnGroups = "offlineSegmentIntervalChecker")
- public void testRealtimeSegmentValidationManager(ITestContext 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();
+ 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");
+ 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);
+ 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);
+ Assert.assertTrue(validationMetrics
+ .getValueOfGauge(ValidationMetrics.makeGaugeName(getDefaultRealtimeTableName(), "TotalDocumentCount")) > 0);
}
// TODO: tests for other ControllerPeriodicTasks (RetentionManagert , RealtimeSegmentValidationManager)
@@ -565,7 +583,8 @@ public class ControllerPeriodicTasksIntegrationTests extends BaseClusterIntegrat
* @throws Exception
*/
@AfterClass
- public void tearDown() throws Exception {
+ public void tearDown()
+ throws Exception {
stopServer();
stopBroker();
stopController();
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
index dce6a5e..321e737 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
@@ -136,17 +136,17 @@ public class OfflineClusterIntegrationTest extends BaseClusterIntegrationTestSet
}
private void registerCallbackHandlers() {
- List<String> instances = _helixAdmin.getInstancesInCluster(_clusterName);
+ List<String> instances = _helixAdmin.getInstancesInCluster(_helixClusterName);
instances.removeIf(instance -> (!instance.startsWith(CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE) && !instance
.startsWith(CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE)));
- List<String> resourcesInCluster = _helixAdmin.getResourcesInCluster(_clusterName);
+ List<String> resourcesInCluster = _helixAdmin.getResourcesInCluster(_helixClusterName);
resourcesInCluster.removeIf(
resource -> (!TableNameBuilder.isTableResource(resource) && !CommonConstants.Helix.BROKER_RESOURCE_INSTANCE
.equals(resource)));
for (String instance : instances) {
List<String> resourcesToMonitor = new ArrayList<>();
for (String resourceName : resourcesInCluster) {
- IdealState idealState = _helixAdmin.getResourceIdealState(_clusterName, resourceName);
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, resourceName);
for (String partitionName : idealState.getPartitionSet()) {
if (idealState.getInstanceSet(partitionName).contains(instance)) {
resourcesToMonitor.add(resourceName);
@@ -155,9 +155,9 @@ public class OfflineClusterIntegrationTest extends BaseClusterIntegrationTestSet
}
}
_serviceStatusCallbacks.add(new ServiceStatus.MultipleCallbackServiceStatusCallback(ImmutableList
- .of(new ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager, _clusterName,
+ .of(new ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager, _helixClusterName,
instance, resourcesToMonitor, 100.0),
- new ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager, _clusterName,
+ new ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager, _helixClusterName,
instance, resourcesToMonitor, 100.0))));
}
}
@@ -741,11 +741,11 @@ public class OfflineClusterIntegrationTest extends BaseClusterIntegrationTestSet
// TODO: Add test to delete broker instance. Currently, stopBroker() does not work correctly.
// Check if '/INSTANCES/<serverName>' has been erased correctly
- String instancePath = "/" + _clusterName + "/INSTANCES/" + serverName;
+ String instancePath = "/" + _helixClusterName + "/INSTANCES/" + serverName;
assertFalse(_propertyStore.exists(instancePath, 0));
// Check if '/CONFIGS/PARTICIPANT/<serverName>' has been erased correctly
- String configPath = "/" + _clusterName + "/CONFIGS/PARTICIPANT/" + serverName;
+ String configPath = "/" + _helixClusterName + "/CONFIGS/PARTICIPANT/" + serverName;
assertFalse(_propertyStore.exists(configPath, 0));
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PinotURIUploadIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PinotURIUploadIntegrationTest.java
index 970100a..f875aaa 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PinotURIUploadIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PinotURIUploadIntegrationTest.java
@@ -195,7 +195,7 @@ public class PinotURIUploadIntegrationTest extends BaseClusterIntegrationTestSet
String serverInstanceId = "Server_localhost_" + CommonConstants.Helix.DEFAULT_SERVER_NETTY_PORT;
// Disable server instance.
- _helixAdmin.enableInstance(getHelixClusterName(), serverInstanceId, false);
+ _helixAdmin.enableInstance(_helixClusterName, serverInstanceId, false);
final String segment6 = "segmentToBeRefreshed_6";
final int nRows1 = 69;
@@ -208,7 +208,7 @@ public class PinotURIUploadIntegrationTest extends BaseClusterIntegrationTestSet
}
// Re-enable the server instance.
- _helixAdmin.enableInstance(getHelixClusterName(), serverInstanceId, true);
+ _helixAdmin.enableInstance(_helixClusterName, serverInstanceId, true);
try {
uploadSegmentsDirectly(segmentTarDir);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java
index 560bf5a..54dacc5 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java
@@ -97,14 +97,14 @@ public class SegmentCompletionIntegrationTest extends BaseClusterIntegrationTest
// Create server instance with the fake server state model
_serverHelixManager = HelixManagerFactory
- .getZKHelixManager(_clusterName, _serverInstance, InstanceType.PARTICIPANT, ZkStarter.DEFAULT_ZK_STR);
+ .getZKHelixManager(_helixClusterName, _serverInstance, InstanceType.PARTICIPANT, ZkStarter.DEFAULT_ZK_STR);
_serverHelixManager.getStateMachineEngine()
.registerStateModelFactory(SegmentOnlineOfflineStateModelFactory.getStateModelName(),
new FakeServerSegmentStateModelFactory());
_serverHelixManager.connect();
// Add Helix tag to the server
- _serverHelixManager.getClusterManagmentTool().addInstanceTag(_clusterName, _serverInstance,
+ _serverHelixManager.getClusterManagmentTool().addInstanceTag(_helixClusterName, _serverInstance,
TableNameBuilder.REALTIME.tableNameWithType(TagNameUtils.DEFAULT_TENANT_NAME));
// Initialize controller leader locator
@@ -127,7 +127,7 @@ public class SegmentCompletionIntegrationTest extends BaseClusterIntegrationTest
@Override
public Boolean apply(@Nullable Void aVoid) {
try {
- ExternalView externalView = _helixAdmin.getResourceExternalView(_clusterName, realtimeTableName);
+ ExternalView externalView = _helixAdmin.getResourceExternalView(_helixClusterName, realtimeTableName);
Map<String, String> stateMap = externalView.getStateMap(_currentSegment);
return stateMap.get(_serverInstance)
.equals(PinotHelixSegmentOnlineOfflineStateModelGenerator.CONSUMING_STATE);
@@ -152,7 +152,7 @@ public class SegmentCompletionIntegrationTest extends BaseClusterIntegrationTest
@Override
public Boolean apply(@Nullable Void aVoid) {
try {
- ExternalView externalView = _helixAdmin.getResourceExternalView(_clusterName, realtimeTableName);
+ ExternalView externalView = _helixAdmin.getResourceExternalView(_helixClusterName, realtimeTableName);
Map<String, String> stateMap = externalView.getStateMap(_currentSegment);
return stateMap.get(_serverInstance).equals(PinotHelixSegmentOnlineOfflineStateModelGenerator.OFFLINE_STATE);
} catch (Exception e) {
@@ -175,7 +175,7 @@ public class SegmentCompletionIntegrationTest extends BaseClusterIntegrationTest
public Boolean apply(@Nullable Void aVoid) {
try {
if (!_currentSegment.equals(oldSegment)) {
- ExternalView externalView = _helixAdmin.getResourceExternalView(_clusterName, realtimeTableName);
+ ExternalView externalView = _helixAdmin.getResourceExternalView(_helixClusterName, realtimeTableName);
Map<String, String> stateMap = externalView.getStateMap(_currentSegment);
return
stateMap.get(_serverInstance).equals(PinotHelixSegmentOnlineOfflineStateModelGenerator.CONSUMING_STATE)
diff --git a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkRealtimeConsumptionSpeed.java b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkRealtimeConsumptionSpeed.java
index 73b0dfe..e86d041 100644
--- a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkRealtimeConsumptionSpeed.java
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkRealtimeConsumptionSpeed.java
@@ -28,8 +28,8 @@ import java.util.concurrent.TimeUnit;
import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.core.realtime.stream.StreamDataServerStartable;
import org.apache.pinot.integration.tests.ClusterIntegrationTestUtils;
-import org.apache.pinot.tools.KafkaStarterUtils;
import org.apache.pinot.integration.tests.RealtimeClusterIntegrationTest;
+import org.apache.pinot.tools.KafkaStarterUtils;
import org.apache.pinot.util.TestUtils;
@@ -41,7 +41,7 @@ public class BenchmarkRealtimeConsumptionSpeed extends RealtimeClusterIntegratio
private static final int ROW_COUNT = 100_000;
private static final int ROW_COUNT_FOR_SEGMENT_FLUSH = 10_000;
private static final long TIMEOUT_MILLIS = 20 * 60 * 1000L; // Twenty minutes
- private final File _tmpDir = new File("/tmp/" + getHelixClusterName());
+ private final File _tmpDir = new File("/tmp/" + _helixClusterName);
private static final int SEGMENT_COUNT = 1;
private static final Random RANDOM = new Random(123456L);
diff --git a/pinot-perf/src/main/java/org/apache/pinot/perf/RealtimeStressTest.java b/pinot-perf/src/main/java/org/apache/pinot/perf/RealtimeStressTest.java
index 8b00c6b..42b4d77 100644
--- a/pinot-perf/src/main/java/org/apache/pinot/perf/RealtimeStressTest.java
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/RealtimeStressTest.java
@@ -28,9 +28,9 @@ import java.util.concurrent.TimeUnit;
import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.core.realtime.stream.StreamDataServerStartable;
import org.apache.pinot.integration.tests.ClusterIntegrationTestUtils;
-import org.apache.pinot.tools.KafkaStarterUtils;
import org.apache.pinot.integration.tests.OfflineClusterIntegrationTest;
import org.apache.pinot.integration.tests.RealtimeClusterIntegrationTest;
+import org.apache.pinot.tools.KafkaStarterUtils;
import org.apache.pinot.util.TestUtils;
@@ -42,7 +42,7 @@ public class RealtimeStressTest extends RealtimeClusterIntegrationTest {
private static final int MIN_ROW_COUNT = 100_000;
private static final int ROW_COUNT_FOR_SEGMENT_FLUSH = 10_000;
private static final long TIMEOUT_MILLIS = 20 * 60 * 1000L; // Twenty minutes
- private final File _tmpDir = new File("/tmp/" + getHelixClusterName());
+ private final File _tmpDir = new File("/tmp/" + _helixClusterName);
private static final int SEGMENT_COUNT = 1;
private static final Random RANDOM = new Random(123456L);
private static long rowsWritten = 0L;
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
index 6fd2ec6..e69f275 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
@@ -300,10 +300,8 @@ public class HelixServerStarter {
List<String> instanceTags = instanceConfig.getTags();
if (instanceTags == null || instanceTags.size() == 0) {
if (ZKMetadataProvider.getClusterTenantIsolationEnabled(_helixManager.getHelixPropertyStore())) {
- _helixAdmin.addInstanceTag(clusterName, instanceName,
- TableNameBuilder.OFFLINE.tableNameWithType(TagNameUtils.DEFAULT_TENANT_NAME));
- _helixAdmin.addInstanceTag(clusterName, instanceName,
- TableNameBuilder.REALTIME.tableNameWithType(TagNameUtils.DEFAULT_TENANT_NAME));
+ _helixAdmin.addInstanceTag(clusterName, instanceName, TagNameUtils.getOfflineTagForTenant(null));
+ _helixAdmin.addInstanceTag(clusterName, instanceName, TagNameUtils.getRealtimeTagForTenant(null));
} else {
_helixAdmin.addInstanceTag(clusterName, instanceName, UNTAGGED_SERVER_INSTANCE);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org