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:46 UTC
[incubator-pinot] 01/01: Clean up test utils,
disconnect fake instances
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