UpgradeCatalog240 extends AbstractUpgradeCatalog { protected void updateSparkConfigs() throws AmbariException { AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class); Clusters clusters = ambariManagementController.getClusters(); Map<String, Cluster> clusterMap = getCheckedClusterMap(clusters); for (final Cluster cluster : clusterMap.values()) { Config sparkDefaultsConfig = cluster.getDesiredConfigByType("spark-defaults"); if (sparkDefaultsConfig != null) { Map<String, String> updates = new HashMap<>(); Map<String, String> sparkDefaultsProperties = sparkDefaultsConfig.getProperties(); if (sparkDefaultsProperties.containsKey("spark.driver.extraJavaOptions")) { LOG.info("Updating property spark.driver.extraJavaOptions in spark-defaults"); String oldValue = sparkDefaultsProperties.get("spark.driver.extraJavaOptions"); if(oldValue.contains("{{hdp_full_version}}")) { String newValue = oldValue.replace("{{hdp_full_version}}", "{{full_stack_version}}"); updates.put("spark.driver.extraJavaOptions", newValue); } } if (sparkDefaultsProperties.containsKey("spark.yarn.am.extraJavaOptions")) { LOG.info("Updating property spark.yarn.am.extraJavaOptions in spark-defaults"); String oldValue = sparkDefaultsProperties.get("spark.yarn.am.extraJavaOptions"); if(oldValue.contains("{{hdp_full_version}}")) { String newValue = oldValue.replace("{{hdp_full_version}}", "{{full_stack_version}}"); updates.put("spark.yarn.am.extraJavaOptions", newValue); } } if(updates.size() != 0 ) { updateConfigurationPropertiesForCluster(cluster, "spark-defaults", updates, null, true, false); } } Config sparkJavaOptsConfig = cluster.getDesiredConfigByType("spark-javaopts-properties"); if(sparkJavaOptsConfig != null) { Map<String, String> updates = new HashMap<>(); Map<String, String> sparkJavaOptsProperties = sparkJavaOptsConfig.getProperties(); if (sparkJavaOptsProperties.containsKey("content")) { String oldValue = sparkJavaOptsProperties.get("content"); if(oldValue.contains("{{hdp_full_version}}")) { LOG.info("Updating property content in spark-javaopts-properties"); String newValue = oldValue.replace("{{hdp_full_version}}", "{{full_stack_version}}"); updates.put("content", newValue); updateConfigurationPropertiesForCluster(cluster, "spark-javaopts-properties", updates, null, true, false); } } } } } @Inject  UpgradeCatalog240(Injector injector); @Override String getTargetVersion(); @Override String getSourceVersion(); static final String DESIRED_VERSION_COLUMN_NAME; static final String BLUEPRINT_SETTING_TABLE; static final String BLUEPRINT_NAME_COL; static final String SETTING_NAME_COL; static final String SETTING_DATA_COL; static final String ID; static final String BLUEPRINT_TABLE; static final String VIEWINSTANCE_TABLE; static final String SHORT_URL_COLUMN; static final String CLUSTER_HANDLE_COLUMN; static final String REQUESTSCHEDULE_TABLE; static final String AUTHENTICATED_USER_ID_COLUMN; static final String TIMELINE_METRICS_SINK_COLLECTION_PERIOD; static final String ONE_DIR_PER_PARITION_PROPERTY; static final String VIEWURL_TABLE; static final String URL_ID_COLUMN; static final String PRECISION_TABLE_TTL_PROPERTY; static final String CLUSTER_SECOND_TABLE_TTL_PROPERTY; static final String CLUSTER_TYPE_COLUMN; static final String REMOTE_AMBARI_CLUSTER_TABLE; static final String REMOTE_AMBARI_CLUSTER_SERVICE_TABLE; static final String CLUSTER_ID; static final String SERVICE_NAME; static final String CLUSTER_NAME; }
@Test public void testSparkConfigUpdate() throws Exception{ Map<String, String> oldPropertiesSparkDefaults = new HashMap<String, String>() { { put("spark.driver.extraJavaOptions", "-Dhdp.version={{hdp_full_version}}"); put("spark.yarn.am.extraJavaOptions", "-Dhdp.version={{hdp_full_version}}"); } }; Map<String, String> newPropertiesSparkDefaults = new HashMap<String, String>() { { put("spark.driver.extraJavaOptions", "-Dhdp.version={{full_stack_version}}"); put("spark.yarn.am.extraJavaOptions", "-Dhdp.version={{full_stack_version}}"); } }; Map<String, String> oldPropertiesSparkJavaOpts = new HashMap<String, String>() { { put("content", "-Dhdp.version={{hdp_full_version}}"); } }; Map<String, String> newPropertiesSparkJavaOpts = new HashMap<String, String>() { { put("content", "-Dhdp.version={{full_stack_version}}"); } }; EasyMockSupport easyMockSupport = new EasyMockSupport(); Clusters clusters = easyMockSupport.createNiceMock(Clusters.class); final Cluster cluster = easyMockSupport.createNiceMock(Cluster.class); Config mockSparkDefaults = easyMockSupport.createNiceMock(Config.class); Config mockSparkJavaOpts = easyMockSupport.createNiceMock(Config.class); expect(clusters.getClusters()).andReturn(new HashMap<String, Cluster>() {{ put("normal", cluster); }}).once(); expect(cluster.getDesiredConfigByType("spark-defaults")).andReturn(mockSparkDefaults).atLeastOnce(); expect(mockSparkDefaults.getProperties()).andReturn(oldPropertiesSparkDefaults).anyTimes(); expect(cluster.getDesiredConfigByType("spark-javaopts-properties")).andReturn(mockSparkJavaOpts).atLeastOnce(); expect(mockSparkJavaOpts.getProperties()).andReturn(oldPropertiesSparkJavaOpts).anyTimes(); Injector injector = easyMockSupport.createNiceMock(Injector.class); expect(injector.getInstance(Gson.class)).andReturn(null).anyTimes(); expect(injector.getInstance(MaintenanceStateHelper.class)).andReturn(null).anyTimes(); expect(injector.getInstance(KerberosHelper.class)).andReturn(createNiceMock(KerberosHelper.class)).anyTimes(); replay(injector, clusters, mockSparkDefaults, mockSparkJavaOpts, cluster); AmbariManagementControllerImpl controller = createMockBuilder(AmbariManagementControllerImpl.class) .addMockedMethod("createConfiguration") .addMockedMethod("getClusters", new Class[] { }) .addMockedMethod("createConfig") .withConstructor(createNiceMock(ActionManager.class), clusters, injector) .createNiceMock(); Injector injector2 = easyMockSupport.createNiceMock(Injector.class); Capture<Map<String, String>> propertiesSparkDefaultsCapture = EasyMock.newCapture(); Capture<Map<String, String>> propertiesSparkJavaOptsCapture = EasyMock.newCapture(); expect(injector2.getInstance(AmbariManagementController.class)).andReturn(controller).anyTimes(); expect(controller.getClusters()).andReturn(clusters).anyTimes(); expect(controller.createConfig(anyObject(Cluster.class), eq("spark-defaults"), capture(propertiesSparkDefaultsCapture), anyString(), EasyMock.<Map<String, Map<String, String>>>anyObject())).andReturn(createNiceMock(Config.class)).once(); expect(controller.createConfig(anyObject(Cluster.class), eq("spark-javaopts-properties"), capture(propertiesSparkJavaOptsCapture), anyString(), EasyMock.<Map<String, Map<String, String>>>anyObject())).andReturn(createNiceMock(Config.class)).once(); replay(controller, injector2); new UpgradeCatalog240(injector2).updateSparkConfigs(); easyMockSupport.verifyAll(); Map<String, String> updatedSparkDefaultsProperties = propertiesSparkDefaultsCapture.getValue(); assertTrue(Maps.difference(newPropertiesSparkDefaults, updatedSparkDefaultsProperties).areEqual()); Map<String, String> updatedSparkJavaOptsProperties = propertiesSparkJavaOptsCapture.getValue(); assertTrue(Maps.difference(newPropertiesSparkJavaOpts, updatedSparkJavaOptsProperties).areEqual()); }