| package org.apache.helix.integration; |
| |
| /* |
| * 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. |
| */ |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeMap; |
| |
| import com.google.common.collect.Maps; |
| import org.apache.helix.HelixDataAccessor; |
| import org.apache.helix.HelixManager; |
| import org.apache.helix.HelixManagerFactory; |
| import org.apache.helix.InstanceType; |
| import org.apache.helix.PropertyType; |
| import org.apache.helix.TestHelper; |
| import org.apache.helix.ZkUnitTestBase; |
| import org.apache.helix.customizedstate.CustomizedStateProvider; |
| import org.apache.helix.customizedstate.CustomizedStateProviderFactory; |
| import org.apache.helix.integration.manager.ClusterControllerManager; |
| import org.apache.helix.integration.manager.MockParticipantManager; |
| import org.apache.helix.model.CustomizedState; |
| import org.apache.helix.model.CustomizedStateConfig; |
| import org.apache.helix.model.CustomizedView; |
| import org.apache.helix.spectator.RoutingTableProvider; |
| import org.apache.helix.spectator.RoutingTableSnapshot; |
| import org.apache.helix.zookeeper.datamodel.ZNRecord; |
| import org.testng.Assert; |
| import org.testng.annotations.AfterClass; |
| import org.testng.annotations.BeforeClass; |
| import org.testng.annotations.Test; |
| |
| |
| public class TestCustomizedViewAggregation extends ZkUnitTestBase { |
| |
| private static CustomizedStateProvider _customizedStateProvider_participant0; |
| private static CustomizedStateProvider _customizedStateProvider_participant1; |
| private static RoutingTableProvider _routingTableProvider; |
| private static HelixManager _spectator; |
| private static HelixManager _manager; |
| // 1st key: customized state type, 2nd key: resource name, 3rd key: partition name, 4th key: instance name, value: state value |
| // This map contains all the customized state information that is updated to ZooKeeper |
| private static Map<String, Map<String, Map<String, Map<String, String>>>> _localCustomizedView; |
| // The set contains customized state types that are enabled for aggregation in config |
| private static Set<String> _aggregationEnabledTypes; |
| // The set contains customized state types that routing table provider shows to users |
| private static Set<String> _routingTableProviderDataSources; |
| private String INSTANCE_0; |
| private String INSTANCE_1; |
| private final String RESOURCE_0 = "TestDB0"; |
| private final String RESOURCE_1 = "TestDB1"; |
| private final String PARTITION_00 = "TestDB0_0"; |
| private final String PARTITION_01 = "TestDB0_1"; |
| private final String PARTITION_10 = "TestDB1_0"; |
| private final String PARTITION_11 = "TestDB1_1"; |
| private MockParticipantManager[] _participants; |
| private ClusterControllerManager _controller; |
| // Customized state values used for test, TYPE_A_0 - TYPE_A_2 are values for Customized state TypeA, etc. |
| private enum CurrentStateValues { |
| TYPE_A_0, TYPE_A_1, TYPE_A_2, TYPE_B_0, TYPE_B_1, TYPE_B_2, TYPE_C_0, TYPE_C_1, TYPE_C_2 |
| } |
| |
| private enum CustomizedStateType { |
| TYPE_A, TYPE_B, TYPE_C |
| } |
| |
| @BeforeClass |
| public void beforeClass() throws Exception { |
| super.beforeClass(); |
| |
| String clusterName = TestHelper.getTestClassName(); |
| int n = 2; |
| |
| System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis())); |
| |
| TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port |
| "localhost", // participant name prefix |
| "TestDB", // resource name prefix |
| 2, // resources |
| 2, // partitions per resource |
| n, // number of nodes |
| 2, // replicas |
| "MasterSlave", true); // do rebalance |
| |
| _controller = |
| new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0"); |
| _controller.syncStart(); |
| |
| // start participants |
| _participants = new MockParticipantManager[n]; |
| for (int i = 0; i < n; i++) { |
| String instanceName = "localhost_" + (12918 + i); |
| |
| _participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName); |
| _participants[i].syncStart(); |
| } |
| |
| INSTANCE_0 = _participants[0].getInstanceName(); |
| INSTANCE_1 = _participants[1].getInstanceName(); |
| |
| _manager = HelixManagerFactory |
| .getZKHelixManager(clusterName, "admin", InstanceType.ADMINISTRATOR, ZK_ADDR); |
| _manager.connect(); |
| |
| _spectator = HelixManagerFactory |
| .getZKHelixManager(clusterName, "spectator", InstanceType.SPECTATOR, ZK_ADDR); |
| _spectator.connect(); |
| |
| // Initialize customized state provider |
| _customizedStateProvider_participant0 = CustomizedStateProviderFactory.getInstance() |
| .buildCustomizedStateProvider(_manager, _participants[0].getInstanceName()); |
| _customizedStateProvider_participant1 = CustomizedStateProviderFactory.getInstance() |
| .buildCustomizedStateProvider(_manager, _participants[1].getInstanceName()); |
| |
| _localCustomizedView = new HashMap<>(); |
| _routingTableProviderDataSources = new HashSet<>(); |
| _aggregationEnabledTypes = new HashSet<>(); |
| |
| List<String> customizedStateTypes = Arrays |
| .asList(CustomizedStateType.TYPE_A.name(), CustomizedStateType.TYPE_B.name(), |
| CustomizedStateType.TYPE_C.name()); |
| |
| CustomizedStateConfig.Builder customizedStateConfigBuilder = |
| new CustomizedStateConfig.Builder(); |
| customizedStateConfigBuilder.setAggregationEnabledTypes(customizedStateTypes); |
| HelixDataAccessor accessor = _manager.getHelixDataAccessor(); |
| accessor.setProperty(accessor.keyBuilder().customizedStateConfig(), |
| customizedStateConfigBuilder.build()); |
| _aggregationEnabledTypes.addAll(customizedStateTypes); |
| |
| Map<PropertyType, List<String>> dataSource = new HashMap<>(); |
| dataSource.put(PropertyType.CUSTOMIZEDVIEW, customizedStateTypes); |
| _routingTableProvider = new RoutingTableProvider(_spectator, dataSource); |
| _routingTableProviderDataSources.addAll(customizedStateTypes); |
| } |
| |
| @AfterClass |
| public void afterClass() { |
| _controller.syncStop(); |
| for (MockParticipantManager participant : _participants) { |
| participant.syncStop(); |
| } |
| _routingTableProvider.shutdown(); |
| _manager.disconnect(); |
| _spectator.disconnect(); |
| } |
| |
| /** |
| * Compare the customized view values between ZK and local record |
| * @throws Exception thread interrupted exception |
| */ |
| private void validateAggregationSnapshot() throws Exception { |
| boolean result = TestHelper.verify(new TestHelper.Verifier() { |
| @Override |
| public boolean verify() { |
| Map<String, Map<String, RoutingTableSnapshot>> routingTableSnapshots = |
| _routingTableProvider.getRoutingTableSnapshots(); |
| |
| // Get customized view snapshot |
| Map<String, RoutingTableSnapshot> fullCustomizedViewSnapshot = |
| routingTableSnapshots.get(PropertyType.CUSTOMIZEDVIEW.name()); |
| |
| if (fullCustomizedViewSnapshot.isEmpty() && !_routingTableProviderDataSources.isEmpty()) { |
| return false; |
| } |
| |
| for (String customizedStateType : fullCustomizedViewSnapshot.keySet()) { |
| if (!_routingTableProviderDataSources.contains(customizedStateType)) { |
| return false; |
| } |
| |
| // Get per customized state type snapshot |
| RoutingTableSnapshot customizedViewSnapshot = |
| fullCustomizedViewSnapshot.get(customizedStateType); |
| |
| // local per customized state type map |
| Map<String, Map<String, Map<String, String>>> localSnapshot = |
| _localCustomizedView.getOrDefault(customizedStateType, Maps.newHashMap()); |
| |
| Collection<CustomizedView> customizedViews = customizedViewSnapshot.getCustomizeViews(); |
| |
| // If a customized state is not set to be aggregated in config, but is enabled in routing table provider, it will show up in customized view returned to user, but will be empty |
| if (!_aggregationEnabledTypes.contains(customizedStateType) |
| && customizedViews.size() != 0) { |
| return false; |
| } |
| |
| if (_aggregationEnabledTypes.contains(customizedStateType) |
| && customizedViews.size() != localSnapshot.size()) { |
| return false; |
| } |
| |
| // Get per resource snapshot |
| for (CustomizedView resourceCustomizedView : customizedViews) { |
| ZNRecord record = resourceCustomizedView.getRecord(); |
| Map<String, Map<String, String>> resourceStateMap = record.getMapFields(); |
| |
| // Get local per resource map |
| Map<String, Map<String, String>> localPerResourceCustomizedView = localSnapshot |
| .getOrDefault(resourceCustomizedView.getResourceName(), Maps.newHashMap()); |
| |
| if (resourceStateMap.size() != localPerResourceCustomizedView.size()) { |
| return false; |
| } |
| |
| // Get per partition snapshot |
| for (String partitionName : resourceStateMap.keySet()) { |
| Map<String, String> stateMap = |
| resourceStateMap.getOrDefault(partitionName, Maps.newTreeMap()); |
| |
| // Get local per partition map |
| Map<String, String> localStateMap = |
| localPerResourceCustomizedView.getOrDefault(partitionName, Maps.newTreeMap()); |
| |
| if (stateMap.isEmpty() && !localStateMap.isEmpty()) { |
| return false; |
| } |
| |
| for (String instanceName : stateMap.keySet()) { |
| // Per instance value |
| String stateMapValue = stateMap.get(instanceName); |
| String localStateMapValue = localStateMap.get(instanceName); |
| if (!stateMapValue.equals(localStateMapValue)) { |
| return false; |
| } |
| } |
| } |
| } |
| } |
| return true; |
| } |
| }, TestHelper.WAIT_DURATION); |
| |
| Assert.assertTrue(result); |
| } |
| |
| /** |
| * Update the local record of customized view |
| * @param instanceName the instance to be updated |
| * @param customizedStateType the customized state type to be updated |
| * @param resourceName the resource to be updated |
| * @param partitionName the partition to be updated |
| * @param customizedStateValue if update, this will be the value to update; a null value indicate delete operation |
| */ |
| private void updateLocalCustomizedViewMap(String instanceName, |
| CustomizedStateType customizedStateType, String resourceName, String partitionName, |
| CurrentStateValues customizedStateValue) { |
| _localCustomizedView.putIfAbsent(customizedStateType.name(), new TreeMap<>()); |
| Map<String, Map<String, Map<String, String>>> localPerStateType = |
| _localCustomizedView.get(customizedStateType.name()); |
| localPerStateType.putIfAbsent(resourceName, new TreeMap<>()); |
| Map<String, Map<String, String>> localPerResource = localPerStateType.get(resourceName); |
| localPerResource.putIfAbsent(partitionName, new TreeMap<>()); |
| Map<String, String> localPerPartition = localPerResource.get(partitionName); |
| if (customizedStateValue == null) { |
| localPerPartition.remove(instanceName); |
| if (localPerPartition.isEmpty()) { |
| localPerResource.remove(partitionName); |
| } |
| } else { |
| localPerPartition.put(instanceName, customizedStateValue.name()); |
| } |
| } |
| |
| /** |
| * Call this method in the test for an update on customized view in both ZK and local map |
| * @param instanceName the instance to be updated |
| * @param customizedStateType the customized state type to be updated |
| * @param resourceName the resource to be updated |
| * @param partitionName the partition to be updated |
| * @param customizedStateValue if update, this will be the value to update; a null value indicate delete operation |
| * @throws Exception if the input instance name is not valid |
| */ |
| private void update(String instanceName, CustomizedStateType customizedStateType, |
| String resourceName, String partitionName, CurrentStateValues customizedStateValue) |
| throws Exception { |
| if (instanceName.equals(INSTANCE_0)) { |
| _customizedStateProvider_participant0 |
| .updateCustomizedState(customizedStateType.name(), resourceName, partitionName, |
| customizedStateValue.name()); |
| updateLocalCustomizedViewMap(INSTANCE_0, customizedStateType, resourceName, partitionName, |
| customizedStateValue); |
| } else if (instanceName.equals(INSTANCE_1)) { |
| _customizedStateProvider_participant1 |
| .updateCustomizedState(customizedStateType.name(), resourceName, partitionName, |
| customizedStateValue.name()); |
| updateLocalCustomizedViewMap(INSTANCE_1, customizedStateType, resourceName, partitionName, |
| customizedStateValue); |
| } else { |
| throw new Exception("The input instance name is not valid."); |
| } |
| } |
| |
| /** |
| * |
| * Call this method in the test for an delete on customized view in both ZK and local map |
| * @param instanceName the instance to be updated |
| * @param customizedStateType the customized state type to be updated |
| * @param resourceName the resource to be updated |
| * @param partitionName the partition to be updated |
| * @throws Exception if the input instance name is not valid |
| */ |
| private void delete(String instanceName, CustomizedStateType customizedStateType, |
| String resourceName, String partitionName) throws Exception { |
| if (instanceName.equals(INSTANCE_0)) { |
| _customizedStateProvider_participant0 |
| .deletePerPartitionCustomizedState(customizedStateType.name(), resourceName, |
| partitionName); |
| updateLocalCustomizedViewMap(INSTANCE_0, customizedStateType, resourceName, partitionName, |
| null); |
| } else if (instanceName.equals(INSTANCE_1)) { |
| _customizedStateProvider_participant1 |
| .deletePerPartitionCustomizedState(customizedStateType.name(), resourceName, |
| partitionName); |
| updateLocalCustomizedViewMap(INSTANCE_1, customizedStateType, resourceName, partitionName, |
| null); |
| } else { |
| throw new Exception("The input instance name is not valid."); |
| } |
| } |
| |
| /** |
| * Set the data sources (customized state types) for routing table provider |
| * @param customizedStateTypes list of customized state types that routing table provider will include in the snapshot shown to users |
| */ |
| |
| /** |
| * Set the customized view aggregation config in controller |
| * @param aggregationEnabledTypes list of customized state types that the controller will aggregate to customized view |
| */ |
| private void setAggregationEnabledTypes(List<CustomizedStateType> aggregationEnabledTypes) { |
| List<String> enabledTypes = new ArrayList<>(); |
| _aggregationEnabledTypes.clear(); |
| for (CustomizedStateType type : aggregationEnabledTypes) { |
| enabledTypes.add(type.name()); |
| _aggregationEnabledTypes.add(type.name()); |
| } |
| CustomizedStateConfig.Builder customizedStateConfigBuilder = |
| new CustomizedStateConfig.Builder(); |
| customizedStateConfigBuilder.setAggregationEnabledTypes(enabledTypes); |
| HelixDataAccessor accessor = _manager.getHelixDataAccessor(); |
| accessor.setProperty(accessor.keyBuilder().customizedStateConfig(), |
| customizedStateConfigBuilder.build()); |
| } |
| |
| @Test |
| public void testCustomizedViewAggregation() throws Exception { |
| |
| // Aggregating: Type A, Type B, Type C |
| // Routing table: Type A, Type B, Type C |
| |
| update(INSTANCE_0, CustomizedStateType.TYPE_A, RESOURCE_0, PARTITION_00, |
| CurrentStateValues.TYPE_A_0); |
| update(INSTANCE_0, CustomizedStateType.TYPE_B, RESOURCE_0, PARTITION_00, |
| CurrentStateValues.TYPE_B_0); |
| update(INSTANCE_0, CustomizedStateType.TYPE_B, RESOURCE_0, PARTITION_01, |
| CurrentStateValues.TYPE_B_1); |
| update(INSTANCE_0, CustomizedStateType.TYPE_A, RESOURCE_1, PARTITION_11, |
| CurrentStateValues.TYPE_A_1); |
| update(INSTANCE_1, CustomizedStateType.TYPE_C, RESOURCE_0, PARTITION_00, |
| CurrentStateValues.TYPE_C_0); |
| update(INSTANCE_1, CustomizedStateType.TYPE_C, RESOURCE_0, PARTITION_01, |
| CurrentStateValues.TYPE_C_1); |
| update(INSTANCE_1, CustomizedStateType.TYPE_B, RESOURCE_1, PARTITION_10, |
| CurrentStateValues.TYPE_B_2); |
| update(INSTANCE_1, CustomizedStateType.TYPE_C, RESOURCE_1, PARTITION_10, |
| CurrentStateValues.TYPE_C_2); |
| update(INSTANCE_1, CustomizedStateType.TYPE_A, RESOURCE_1, PARTITION_11, |
| CurrentStateValues.TYPE_A_1); |
| validateAggregationSnapshot(); |
| |
| Assert.assertNull(_customizedStateProvider_participant0 |
| .getCustomizedState(CustomizedStateType.TYPE_C.name(), RESOURCE_0)); |
| |
| // Test batch update API to update several customized state fields in the same customized state, but for now only CURRENT_STATE will be aggregated in customized view |
| Map<String, String> customizedStates = Maps.newHashMap(); |
| customizedStates.put("CURRENT_STATE", CurrentStateValues.TYPE_A_2.name()); |
| customizedStates.put("PREVIOUS_STATE", CurrentStateValues.TYPE_A_0.name()); |
| _customizedStateProvider_participant1 |
| .updateCustomizedState(CustomizedStateType.TYPE_A.name(), RESOURCE_1, PARTITION_10, |
| customizedStates); |
| updateLocalCustomizedViewMap(INSTANCE_1, CustomizedStateType.TYPE_A, RESOURCE_1, PARTITION_10, |
| CurrentStateValues.TYPE_A_2); |
| |
| validateAggregationSnapshot(); |
| |
| // Aggregating: Type A |
| // Routing table: Type A, Type B, Type C |
| setAggregationEnabledTypes(Arrays.asList(CustomizedStateType.TYPE_A)); |
| // This is commented out as a work around to pass the test |
| // The validation of config change will be done combined with the next several customized state changes |
| // The next validation should only show TYPE_A states aggregated in customized view |
| // Until we fix the issue in routing table provider https://github.com/apache/helix/issues/1296 |
| // validateAggregationSnapshot(); |
| |
| // Test get customized state and get per partition customized state via customized state provider, this part of test doesn't change customized view |
| CustomizedState customizedState = _customizedStateProvider_participant1 |
| .getCustomizedState(CustomizedStateType.TYPE_A.name(), RESOURCE_1); |
| Assert.assertEquals(customizedState.getState(PARTITION_10), CurrentStateValues.TYPE_A_2.name()); |
| Assert.assertEquals(customizedState.getPreviousState(PARTITION_10), |
| CurrentStateValues.TYPE_A_0.name()); |
| Assert.assertEquals(customizedState.getState(PARTITION_11), CurrentStateValues.TYPE_A_1.name()); |
| Map<String, String> perPartitionCustomizedState = _customizedStateProvider_participant1 |
| .getPerPartitionCustomizedState(CustomizedStateType.TYPE_A.name(), RESOURCE_1, |
| PARTITION_10); |
| // Remove this field because it's automatically updated for monitoring purpose and we don't need to compare it |
| perPartitionCustomizedState.remove(CustomizedState.CustomizedStateProperty.START_TIME.name()); |
| Map<String, String> actualPerPartitionCustomizedState = Maps.newHashMap(); |
| actualPerPartitionCustomizedState |
| .put(CustomizedState.CustomizedStateProperty.CURRENT_STATE.name(), |
| CurrentStateValues.TYPE_A_2.name()); |
| actualPerPartitionCustomizedState |
| .put(CustomizedState.CustomizedStateProperty.PREVIOUS_STATE.name(), |
| CurrentStateValues.TYPE_A_0.name()); |
| Assert.assertEquals(perPartitionCustomizedState, actualPerPartitionCustomizedState); |
| |
| // Test delete per partition customized state via customized state provider. |
| _customizedStateProvider_participant1 |
| .deletePerPartitionCustomizedState(CustomizedStateType.TYPE_A.name(), RESOURCE_1, |
| PARTITION_10); |
| customizedState = _customizedStateProvider_participant1 |
| .getCustomizedState(CustomizedStateType.TYPE_A.name(), RESOURCE_1); |
| Assert.assertEquals(customizedState.getState(PARTITION_11), CurrentStateValues.TYPE_A_1.name()); |
| Assert.assertNull(_customizedStateProvider_participant1 |
| .getPerPartitionCustomizedState(CustomizedStateType.TYPE_A.name(), RESOURCE_1, |
| PARTITION_10)); |
| // Customized view only reflect CURRENT_STATE field |
| updateLocalCustomizedViewMap(INSTANCE_1, CustomizedStateType.TYPE_A, RESOURCE_1, PARTITION_10, |
| null); |
| validateAggregationSnapshot(); |
| |
| // Update some customized states and verify |
| delete(INSTANCE_0, CustomizedStateType.TYPE_A, RESOURCE_0, PARTITION_00); |
| delete(INSTANCE_1, CustomizedStateType.TYPE_B, RESOURCE_1, PARTITION_10); |
| |
| // delete a customize state that does not exist |
| delete(INSTANCE_1, CustomizedStateType.TYPE_A, RESOURCE_1, PARTITION_10); |
| validateAggregationSnapshot(); |
| |
| // Aggregating: Type A, Type B, Type C |
| // Routing table: Type A, Type B, Type C |
| setAggregationEnabledTypes(Arrays.asList(CustomizedStateType.TYPE_A, CustomizedStateType.TYPE_B, |
| CustomizedStateType.TYPE_C)); |
| validateAggregationSnapshot(); |
| |
| update(INSTANCE_0, CustomizedStateType.TYPE_B, RESOURCE_0, PARTITION_01, |
| CurrentStateValues.TYPE_B_2); |
| update(INSTANCE_1, CustomizedStateType.TYPE_B, RESOURCE_1, PARTITION_10, |
| CurrentStateValues.TYPE_B_1); |
| update(INSTANCE_1, CustomizedStateType.TYPE_C, RESOURCE_1, PARTITION_10, |
| CurrentStateValues.TYPE_C_0); |
| update(INSTANCE_0, CustomizedStateType.TYPE_A, RESOURCE_1, PARTITION_11, |
| CurrentStateValues.TYPE_A_0); |
| validateAggregationSnapshot(); |
| } |
| } |
