You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by vb...@apache.org on 2016/08/02 13:05:24 UTC
ambari git commit: AMBARI-17979. YARN restart icon appeared after
long time after deleting ZK server.(vbrodetskyi)
Repository: ambari
Updated Branches:
refs/heads/trunk a8c0432a9 -> 935266df3
AMBARI-17979. YARN restart icon appeared after long time after deleting ZK server.(vbrodetskyi)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/935266df
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/935266df
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/935266df
Branch: refs/heads/trunk
Commit: 935266df3722a576a2596d724db9d12644e3a763
Parents: a8c0432
Author: Vitaly Brodetskyi <vb...@hortonworks.com>
Authored: Tue Aug 2 16:04:55 2016 +0300
Committer: Vitaly Brodetskyi <vb...@hortonworks.com>
Committed: Tue Aug 2 16:04:55 2016 +0300
----------------------------------------------------------------------
.../ambari/server/agent/HeartBeatHandler.java | 2 -
.../server/configuration/Configuration.java | 2 +-
.../internal/ConfigGroupResourceProvider.java | 1 -
.../ambari/server/state/ConfigHelper.java | 197 +++----------------
.../apache/ambari/server/state/HostConfig.java | 6 +
.../server/state/cluster/ClusterImpl.java | 4 -
.../svccomphost/ServiceComponentHostImpl.java | 1 -
.../ConfigGroupResourceProviderTest.java | 3 -
.../ambari/server/state/ConfigHelperTest.java | 137 ++-----------
.../svccomphost/ServiceComponentHostTest.java | 70 -------
10 files changed, 48 insertions(+), 375 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
index f9ce8e6..a25b875 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java
@@ -473,8 +473,6 @@ public class HeartBeatHandler {
now));
}
- configHelper.invalidateStaleConfigsCache(hostname, null);
-
response.setStatusCommands(cmds);
response.setResponseStatus(RegistrationStatus.OK);
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java b/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
index 6e3eef7..fae9378 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java
@@ -473,7 +473,7 @@ public class Configuration {
private static final long SERVER_EC_CACHE_SIZE_DEFAULT = 10000L;
private static final String SERVER_STALE_CONFIG_CACHE_ENABLED_DEFAULT = "true";
- private static final String SERVER_STALE_CONFIG_CACHE_EXPIRATION_DEFAULT = "300";
+ private static final String SERVER_STALE_CONFIG_CACHE_EXPIRATION_DEFAULT = "600";
private static final String SERVER_JDBC_USER_NAME_DEFAULT = "ambari";
private static final String SERVER_JDBC_USER_PASSWD_DEFAULT = "bigdata";
private static final String SERVER_JDBC_RCA_USER_NAME_DEFAULT = "mapred";
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProvider.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProvider.java
index 3ba8e69..96bb8f9 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProvider.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProvider.java
@@ -696,7 +696,6 @@ public class ConfigGroupResourceProvider extends
LOG.warn("Could not determine service name for config group {}, service config version not created",
configGroup.getId());
}
- getManagementController().getConfigHelper().invalidateStaleConfigsCache(cluster.getDesiredConfigs());
}
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/main/java/org/apache/ambari/server/state/ConfigHelper.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/ConfigHelper.java b/ambari-server/src/main/java/org/apache/ambari/server/state/ConfigHelper.java
index 16edc74..70c24f9 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/ConfigHelper.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/ConfigHelper.java
@@ -29,12 +29,9 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import com.google.common.base.Objects;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.configuration.Configuration;
@@ -51,7 +48,6 @@ import org.slf4j.LoggerFactory;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Maps;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;
@@ -71,25 +67,10 @@ public class ConfigHelper {
private final int STALE_CONFIGS_CACHE_EXPIRATION_TIME;
/**
- * This lock used to synchronize {@link #staleConfigCacheDesiredConfigs} variable writing and avoid
- * {@link #staleConfigsCache} invalidation during stale config reading(it will be invalidated when all reading
- * operations are finished, thus next updating of {@link #staleConfigsCache} will be performed with fresh
- * {@link #staleConfigCacheDesiredConfigs}
+ * Cache for storing stale config flags. Key for cache is hash of [actualConfigs, desiredConfigs, hostName, serviceName,
+ * componentName].
*/
- private final ReentrantReadWriteLock staleConfigCacheLock = new ReentrantReadWriteLock();
- /**
- * A cache of which {@link ServiceComponentHost}s have stale configurations.
- * This cache may be invalidated within the context of a running transaction
- * which could potentially cause old data to be re-cached before the
- * transaction has completed.
- * <p/>
- * As a result, all invalidation of this cache should be done on a separate
- * thread using the {@link ConfigHelper#staleConfigCacheLock} with desiredConfigs
- * from thread that requested invalidation.
- *
- * @see #cacheInvalidationExecutor
- */
- private final Cache<ServiceComponentHost, Boolean> staleConfigsCache;
+ private final Cache<Integer, Boolean> staleConfigsCache;
private static final Logger LOG =
LoggerFactory.getLogger(ConfigHelper.class);
@@ -118,24 +99,6 @@ public class ConfigHelper {
*/
public static final String FIRST_VERSION_TAG = "version1";
- /**
- * A {@link ThreadFactory} for the {@link #cacheInvalidationExecutor}.
- */
- private final ThreadFactory cacheInvalidationThreadFactory = new ThreadFactoryBuilder().setNameFormat(
- "ambari-stale-config-invalidator-%d").setDaemon(true).build();
-
- /**
- * Used to invalidate the {@link #staleConfigsCache} on a separate thread.
- */
- private final ExecutorService cacheInvalidationExecutor = createCacheInvalidationExecutor();
-
- /**
- * Every request that updates configs or configs-groups must invalidate staleConfigCache and update this filed
- * with its own updated desiredConfigs(they must be passed form updater request) to avoid any issues with EL caches
- * during parallel requests.
- */
- private volatile Map<String, DesiredConfig> staleConfigCacheDesiredConfigs = null;
-
@Inject
public ConfigHelper(Clusters c, AmbariMetaInfo metaInfo, Configuration configuration, ClusterDAO clusterDAO) {
clusters = c;
@@ -472,74 +435,12 @@ public class ConfigHelper {
*/
public boolean isStaleConfigs(ServiceComponentHost sch, Map<String, DesiredConfig> requestDesiredConfigs)
throws AmbariException {
- Boolean stale = null;
- staleConfigCacheLock.readLock().lock();
- try {
- // try the cache
- if (STALE_CONFIGS_CACHE_ENABLED) {
- stale = staleConfigsCache.getIfPresent(sch);
- }
-
- if (stale == null) {
- /*
- prefer staleConfigCacheDesiredConfigs(they are supposed to be passed from thread that updated configs) to avoid
- population staleConfigsCache with stale data in case current thread have stale EL caches(this can happen during
- parallel requests)
- */
- Map<String, DesiredConfig> desiredConfigs = staleConfigCacheDesiredConfigs != null ? staleConfigCacheDesiredConfigs : requestDesiredConfigs;
- stale = calculateIsStaleConfigs(sch, desiredConfigs);
- staleConfigsCache.put(sch, stale);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Cache configuration staleness for host {} and component {} as {}",
- sch.getHostName(), sch.getServiceComponentName(), stale);
- }
- }
- return stale;
- } finally {
- staleConfigCacheLock.readLock().unlock();
+ boolean stale = calculateIsStaleConfigs(sch, requestDesiredConfigs);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cache configuration staleness for host {} and component {} as {}",
+ sch.getHostName(), sch.getServiceComponentName(), stale);
}
-
- }
-
- /**
- * Invalidates the stale configuration cache for all
- * {@link ServiceComponentHost}s for the given host. This will execute the
- * invalidation on a separate thread that will try to get write lock from {@link ConfigHelper#staleConfigCacheLock} to
- * wait for finish of all reading operations for stale configs. Every subsequent read operation will be blocked with
- * write lock and will be executed only after {@link #staleConfigCacheDesiredConfigs} updated and cache invalidated.
- *
- * @param hostname
- * @param desiredConfigs desired cluster configs that will be used for configuration invalidation
- */
- public void invalidateStaleConfigsCache(String hostname, Map<String, DesiredConfig> desiredConfigs) {
- try {
- for (Cluster cluster : clusters.getClustersForHost(hostname)) {
- List<ServiceComponentHost> serviceComponentHosts = cluster.getServiceComponentHosts(hostname);
- Runnable invalidationRunnable = new StaleConfigInvalidationRunnable(serviceComponentHosts, desiredConfigs);
- cacheInvalidationExecutor.execute(invalidationRunnable);
- }
- } catch (AmbariException e) {
- LOG.warn("Unable to find clusters for host " + hostname);
- }
- }
-
- /**
- * Invalidates the stale configuration cache for keys. This will execute the
- * invalidation on a separate thread.
- */
- public void invalidateStaleConfigsCache(Map<String, DesiredConfig> desiredConfigs) {
- Runnable invalidationRunnable = new StaleConfigInvalidationRunnable(desiredConfigs);
- cacheInvalidationExecutor.execute(invalidationRunnable);
- }
-
- /**
- * Invalidates the stale configuration cache for a single
- * {@link ServiceComponentHost}. This will execute the invalidation on a
- * separate thread.
- */
- public void invalidateStaleConfigsCache(ServiceComponentHost sch, Map<String, DesiredConfig> desiredConfigs) {
- Runnable invalidationRunnable = new StaleConfigInvalidationRunnable(Collections.singletonList(sch), desiredConfigs);
- cacheInvalidationExecutor.execute(invalidationRunnable);
+ return stale;
}
/**
@@ -1115,11 +1016,28 @@ public class ConfigHelper {
}
Cluster cluster = clusters.getClusterById(sch.getClusterId());
- StackId stackId = cluster.getDesiredStackVersion();
Map<String, Map<String, String>> desired = getEffectiveDesiredTags(cluster, sch.getHostName(),
desiredConfigs);
+ Boolean stale = null;
+ int staleHash = 0;
+ if (STALE_CONFIGS_CACHE_ENABLED){
+ staleHash = Objects.hashCode(actual.hashCode(),
+ desired.hashCode(),
+ sch.getHostName(),
+ sch.getServiceComponentName(),
+ sch.getServiceName());
+ stale = staleConfigsCache.getIfPresent(staleHash);
+ if(stale != null) {
+ return stale;
+ }
+ }
+
+ stale = false;
+
+ StackId stackId = cluster.getDesiredStackVersion();
+
ServiceInfo serviceInfo = ambariMetaInfo.getService(stackId.getStackName(),
stackId.getStackVersion(), sch.getServiceName());
@@ -1132,7 +1050,6 @@ public class ConfigHelper {
// --- desired tags DO match actual tags: not_stale
// --- desired tags DO NOT match actual tags
// ---- merge values, determine changed keys, check stack: stale
- boolean stale = false;
Iterator<Entry<String, Map<String, String>>> it = desired.entrySet().iterator();
@@ -1178,6 +1095,9 @@ public class ConfigHelper {
}
}
}
+ if (STALE_CONFIGS_CACHE_ENABLED) {
+ staleConfigsCache.put(staleHash, stale);
+ }
return stale;
}
@@ -1375,61 +1295,4 @@ public class ConfigHelper {
}
}
- /**
- * Invalidates the {@link ConfigHelper#staleConfigsCache} after acquiring a
- * lock around {@link ConfigHelper#staleConfigCacheLock}. It is necessary to acquire
- * this lock to avoid parallel reading threads populate cache using stale desiredConfigs.
- * This thread can receive desiredConfigs form thread that initialized invalidation and put
- * them to {@link ConfigHelper}. {@link ConfigHelper#staleConfigCacheLock} guaranteeing that invalidation
- * will happen only after desiredConfigs passed to {@link ConfigHelper} instance.
- */
- private final class StaleConfigInvalidationRunnable implements Runnable {
-
- private final List<ServiceComponentHost> m_keysToInvalidate;
- private final Map<String, DesiredConfig> m_desiredConfigs;
-
- /**
- * Constructor.
- *
- */
- private StaleConfigInvalidationRunnable(Map<String, DesiredConfig> desiredConfigs) {
- m_keysToInvalidate = null;
- m_desiredConfigs = desiredConfigs;
- }
-
- /**
- * Constructor.
- *
- * @param keysToInvalidate
- * the keys to invalidate in the cache.
- */
- private StaleConfigInvalidationRunnable(List<ServiceComponentHost> keysToInvalidate, Map<String, DesiredConfig> desiredConfigs) {
- m_keysToInvalidate = keysToInvalidate;
- m_desiredConfigs = desiredConfigs;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void run() {
- staleConfigCacheLock.writeLock().lock();
- try {
- staleConfigCacheDesiredConfigs = m_desiredConfigs;
- if (null == m_keysToInvalidate || m_keysToInvalidate.isEmpty()) {
- staleConfigsCache.invalidateAll();
- } else {
- staleConfigsCache.invalidateAll(m_keysToInvalidate);
- }
- } finally {
- staleConfigCacheLock.writeLock().unlock();
- }
- }
- }
-
- private ExecutorService createCacheInvalidationExecutor() {
- return Executors.newSingleThreadExecutor(
- cacheInvalidationThreadFactory);
- }
-
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/main/java/org/apache/ambari/server/state/HostConfig.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/HostConfig.java b/ambari-server/src/main/java/org/apache/ambari/server/state/HostConfig.java
index 0f451ce..fc22ba5 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/HostConfig.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/HostConfig.java
@@ -17,6 +17,7 @@
*/
package org.apache.ambari.server.state;
+import com.google.common.base.Objects;
import org.codehaus.jackson.annotate.JsonProperty;
import org.codehaus.jackson.map.annotate.JsonSerialize;
@@ -51,6 +52,11 @@ public class HostConfig {
}
@Override
+ public int hashCode(){
+ return Objects.hashCode(defaultVersionTag.hashCode(), configGroupOverrides.hashCode());
+ }
+
+ @Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("{");
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
index fb00cf5..0d212a1 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java
@@ -543,7 +543,6 @@ public class ClusterImpl implements Cluster {
+ configGroup.getTag());
} else {
clusterConfigGroups.put(configGroup.getId(), configGroup);
- configHelper.invalidateStaleConfigsCache(getDesiredConfigs());
}
} finally {
@@ -655,7 +654,6 @@ public class ClusterImpl implements Cluster {
configGroup.delete();
clusterConfigGroups.remove(id);
- configHelper.invalidateStaleConfigsCache(getDesiredConfigs());
} finally {
clusterGlobalLock.writeLock().unlock();
}
@@ -2346,7 +2344,6 @@ public class ClusterImpl implements Cluster {
ServiceConfigVersionResponse serviceConfigVersionResponse = applyConfigs(
configs, user, serviceConfigVersionNote);
- configHelper.invalidateStaleConfigsCache(getDesiredConfigs());
return serviceConfigVersionResponse;
} finally {
clusterGlobalLock.writeLock().unlock();
@@ -2565,7 +2562,6 @@ public class ClusterImpl implements Cluster {
try {
ServiceConfigVersionResponse serviceConfigVersionResponse = applyServiceConfigVersion(
serviceName, version, user, note);
- configHelper.invalidateStaleConfigsCache(getDesiredConfigs());
return serviceConfigVersionResponse;
} finally {
clusterGlobalLock.writeLock().unlock();
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
index 78f3751..3b5ed28 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java
@@ -1797,7 +1797,6 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
if (desiredStateEntity != null) {
desiredStateEntity.setRestartRequired(restartRequired);
saveComponentDesiredStateEntityIfPersisted();
- helper.invalidateStaleConfigsCache(this, null);
} else {
LOG.warn("Setting a member on an entity object that may have been " +
"previously deleted, serviceName = " + getServiceName() + ", " +
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProviderTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProviderTest.java b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProviderTest.java
index 2a4b949..4dec990 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProviderTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ConfigGroupResourceProviderTest.java
@@ -514,9 +514,6 @@ public class ConfigGroupResourceProviderTest {
return configGroupMap;
}
});
- expect(managementController.getConfigHelper()).andReturn(configHelper).once();
- configHelper.invalidateStaleConfigsCache(null);
- expectLastCall().once();
replay(managementController, clusters, cluster,
configGroup, response, configGroupResponse, configHelper, hostDAO, hostEntity1, hostEntity2, h1, h2);
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/test/java/org/apache/ambari/server/state/ConfigHelperTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/ConfigHelperTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/ConfigHelperTest.java
index f3d35d9..37a48f0 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/state/ConfigHelperTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/state/ConfigHelperTest.java
@@ -195,19 +195,13 @@ public class ConfigHelperTest {
cr4.setPropertiesAttributes(null);
final ClusterRequest clusterRequest4 =
- new ClusterRequest(cluster.getClusterId(), clusterName,
- cluster.getDesiredStackVersion().getStackVersion(), null);
+ new ClusterRequest(cluster.getClusterId(), clusterName,
+ cluster.getDesiredStackVersion().getStackVersion(), null);
clusterRequest4.setDesiredConfig(Collections.singletonList(cr4));
managementController.updateClusters(new HashSet<ClusterRequest>() {{
add(clusterRequest4);
}}, null);
-
- // instrument the asynchronous stale config invalidation to be synchronous
- // so that tests pass
- Field field = ConfigHelper.class.getDeclaredField("cacheInvalidationExecutor");
- field.setAccessible(true);
- field.set(configHelper, new SynchronousThreadPoolExecutor());
}
@After
@@ -220,7 +214,7 @@ public class ConfigHelperTest {
@Transactional
Long addConfigGroup(String name, String tag, List<String> hosts,
- List<Config> configs) throws AmbariException {
+ List<Config> configs) throws AmbariException {
Map<Long, Host> hostMap = new HashMap<Long, Host>();
Map<String, Config> configMap = new HashMap<String, Config>();
@@ -265,15 +259,15 @@ public class ConfigHelperTest {
}
@Test
- public void testProcessHiddenAttribute() throws Exception{
+ public void testProcessHiddenAttribute() throws Exception {
StackInfo stackInfo = metaInfo.getStack("HDP", "2.0.5");
Map<String, Map<String, Map<String, String>>> configAttributes = new HashMap<String, Map<String, Map<String, String>>>();
configAttributes.put("hive-site", stackInfo.getDefaultConfigAttributesForConfigType("hive-site"));
Map<String, Map<String, String>> originalConfig_hiveClient = createHiveConfig();
- Map<String, Map<String, String>> expectedConfig_hiveClient = new HashMap<String, Map<String, String>>(){{
- put("hive-site", new HashMap<String, String>(){{
+ Map<String, Map<String, String>> expectedConfig_hiveClient = new HashMap<String, Map<String, String>>() {{
+ put("hive-site", new HashMap<String, String>() {{
put("javax.jdo.option.ConnectionDriverName", "oracle");
put("hive.metastore.warehouse.dir", "/tmp");
}});
@@ -296,9 +290,9 @@ public class ConfigHelperTest {
Assert.assertEquals(expectedConfig_hiveServer1, originalConfig_hiveServer1);
}
- private Map<String, Map<String, String>> createHiveConfig(){
- return new HashMap<String, Map<String, String>>(){{
- put("hive-site", new HashMap<String, String>(){{
+ private Map<String, Map<String, String>> createHiveConfig() {
+ return new HashMap<String, Map<String, String>>() {{
+ put("hive-site", new HashMap<String, String>() {{
put("javax.jdo.option.ConnectionDriverName", "oracle");
put("javax.jdo.option.ConnectionPassword", "1");
put("hive.metastore.warehouse.dir", "/tmp");
@@ -666,7 +660,7 @@ public class ConfigHelperTest {
Assert.assertEquals(1, propertiesAttributes.size());
Assert.assertTrue(propertiesAttributes.containsKey("attribute1"));
// Config tag before update
- Assert.assertEquals("version1",currentConfig.getTag());
+ Assert.assertEquals("version1", currentConfig.getTag());
// Properties before update
Assert.assertEquals("30", properties.get("fs.trash.interval"));
// Property and attribute exist
@@ -717,7 +711,7 @@ public class ConfigHelperTest {
updates.put("oozie.service.HadoopAccessorService.kerberos.enabled", "true");
configHelper.updateConfigType(cluster, managementController, "oozie-site", updates, null, "admin", "Test " +
- "note");
+ "note");
Config updatedConfig = cluster.getDesiredConfigByType("oozie-site");
// Config tag updated
@@ -782,7 +776,6 @@ public class ConfigHelperTest {
hc2.setDefaultVersionTag("version1");
schReturn.put("flume-conf", hc2);
// invalidate cache to test new sch
- configHelper.invalidateStaleConfigsCache(null);
// Cluster level same configs
Assert.assertFalse(configHelper.isStaleConfigs(sch, null));
@@ -802,7 +795,6 @@ public class ConfigHelperTest {
hc3.setDefaultVersionTag("version1");
hc3.getConfigGroupOverrides().put(1l, "FLUME1");
schReturn.put("flume-conf", hc3);
- configHelper.invalidateStaleConfigsCache(null);
// version1 and FLUME1 - stale=false
Assert.assertFalse(configHelper.isStaleConfigs(sch, null));
@@ -811,7 +803,6 @@ public class ConfigHelperTest {
hc4.setDefaultVersionTag("version1");
hc4.getConfigGroupOverrides().put(1l, "FLUME2");
schReturn.put("flume-conf", hc4);
- configHelper.invalidateStaleConfigsCache(null);
// version1 and FLUME2 - stale=true
Assert.assertTrue(configHelper.isStaleConfigs(sch, null));
@@ -820,118 +811,12 @@ public class ConfigHelperTest {
hc5.setDefaultVersionTag("version3");
hc5.getConfigGroupOverrides().put(1l, "FLUME1");
schReturn.put("flume-conf", hc5);
- configHelper.invalidateStaleConfigsCache(null);
// version3 and FLUME1 - stale=true
Assert.assertTrue(configHelper.isStaleConfigs(sch, null));
verify(sch);
}
- @Test
- public void testCalculateIsStaleConfigsParallel() throws Exception{
- Map<String, HostConfig> schReturn = new HashMap<String, HostConfig>();
- HostConfig hc = new HostConfig();
- // Put a different version to check for change
- hc.setDefaultVersionTag("version2");
- schReturn.put("flume-conf", hc);
-
- // set up mocks
- final ServiceComponentHost sch = createNiceMock(ServiceComponentHost.class);
- // set up expectations
- expect(sch.getActualConfigs()).andReturn(schReturn).anyTimes();
- expect(sch.getHostName()).andReturn("h1").anyTimes();
- expect(sch.getClusterId()).andReturn(1l).anyTimes();
- expect(sch.getServiceName()).andReturn("FLUME").anyTimes();
- expect(sch.getServiceComponentName()).andReturn("FLUME_HANDLER").anyTimes();
- replay(sch);
- // Cluster level config changes
-
- final Config config1 = cluster.getDesiredConfigByType("flume-conf");
-
- applyConfig(new HashMap<String, String>(){{
- put("property", "1");
- }}, "flume-conf", "version2");
-
- final Config config2 = cluster.getDesiredConfigByType("flume-conf");
-
- applyConfig(new HashMap<String, String>(){{
- put("property", "2");
- }}, "flume-conf", "version3");
-
- final Config config3 = cluster.getDesiredConfigByType("flume-conf");
-
- cluster.addDesiredConfig("admin", new HashSet<Config>(){{add(config1);}});
-
- final AtomicBoolean mustBeStale = new AtomicBoolean();
- mustBeStale.set(false);
-
- final AtomicBoolean failed = new AtomicBoolean();
- failed.set(false);
-
- final AtomicBoolean finished = new AtomicBoolean();
- finished.set(false);
- // avoid situations when not checked previous mustBeStale value yes and applied new config version
- final Lock checkLock = new ReentrantLock();
-
- // parallel thread that will compare actual stale config with expected accordingly to desired configs, checks if
- // isStaleConfigs bypassing every cache and returns correct information.
- Thread parallel = new Thread(new Runnable() {
- @Override
- public void run() {
- while(!finished.get()){
- checkLock.lock();
- try {
- boolean isStale = configHelper.isStaleConfigs(sch, null);
- if(mustBeStale.get() != isStale){
- failed.set(true);
- break;
- }
- } catch (AmbariException e) {
- e.printStackTrace();
- } finally {
- checkLock.unlock();
- }
- }
- }
- });
-
- parallel.start();
-
- Random r = new Random();
- for(int i=0; i< 1000; i++){
- try {
- checkLock.lock();
- switch(r.nextInt(3)) {
- case 0: {
- cluster.addDesiredConfig("admin", new HashSet<Config>(){{add(config1);}});
- mustBeStale.set(true);
- checkLock.unlock();
- break;
- }
- case 1: {
- cluster.addDesiredConfig("admin", new HashSet<Config>(){{add(config2);}});
- mustBeStale.set(false);
- checkLock.unlock();
- break;
- }
- case 2: {
- cluster.addDesiredConfig("admin", new HashSet<Config>(){{add(config3);}});
- mustBeStale.set(true);
- checkLock.unlock();
- break;
- }
- }
- } catch (Exception e){
- checkLock.unlock();
- }
- if(!parallel.isAlive()) {
- break;
- }
- }
- finished.set(true);
- parallel.join();
- Assert.assertFalse(failed.get());
- }
}
public static class RunWithCustomModule {
http://git-wip-us.apache.org/repos/asf/ambari/blob/935266df/ambari-server/src/test/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostTest.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostTest.java
index 44afdaa..0383e4c 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostTest.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostTest.java
@@ -18,9 +18,6 @@
package org.apache.ambari.server.state.svccomphost;
-import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -28,8 +25,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
import javax.persistence.EntityManager;
@@ -795,7 +790,6 @@ public class ServiceComponentHostTest {
makeConfig(cluster, "foo", "version1",
new HashMap<String,String>() {{ put("a", "c"); }}, new HashMap<String, Map<String,String>>());
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS does not define type 'foo', so changes do not count to stale
Assert.assertFalse(sch1.convertToResponse(null).isStaleConfig());
@@ -803,7 +797,6 @@ public class ServiceComponentHostTest {
makeConfig(cluster, "hdfs-site", "version1",
new HashMap<String,String>() {{ put("a", "b"); }}, new HashMap<String, Map<String,String>>());
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site is not on the actual, but it is defined, so it is stale
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
@@ -812,23 +805,11 @@ public class ServiceComponentHostTest {
actual.put("hdfs-site", new HashMap<String, String>() {{ put ("tag", "version1"); }});
sch1.updateActualConfigs(actual);
- // previous value from cache
- Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
- //reset restartRequired flag + invalidating isStale cache
- // after start/restart command execution completed
- sch1.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site up to date, only for sch1
Assert.assertFalse(sch1.convertToResponse(null).isStaleConfig());
Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
sch2.updateActualConfigs(actual);
- // previous value from cache
- Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
- //reset restartRequired flag + invalidating isStale cache(
- // after start/restart command execution completed)
- sch2.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site up to date for both
Assert.assertFalse(sch1.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch2.convertToResponse(null).isStaleConfig());
@@ -836,7 +817,6 @@ public class ServiceComponentHostTest {
makeConfig(cluster, "hdfs-site", "version2",
new HashMap<String, String>() {{ put("dfs.journalnode.http-address", "http://foo"); }},
new HashMap<String, Map<String,String>>());
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site updated to changed property
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
@@ -849,7 +829,6 @@ public class ServiceComponentHostTest {
// after start/restart command execution completed
sch1.setRestartRequired(false);
sch2.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site updated to changed property
Assert.assertFalse(sch1.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch2.convertToResponse(null).isStaleConfig());
@@ -870,7 +849,6 @@ public class ServiceComponentHostTest {
new HashMap<Long, Host>() {{ put(hostEntity.getHostId(), host); }});
configGroup.persist();
cluster.addConfigGroup(configGroup);
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site updated host to changed property
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
@@ -878,23 +856,11 @@ public class ServiceComponentHostTest {
actual.get("hdfs-site").put(configGroup.getId().toString(), "version3");
sch2.updateActualConfigs(actual);
- // previous value from cache
- Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
- //reset restartRequired flag + invalidating isStale cache
- // after start/restart command execution completed
- sch2.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site updated host to changed property
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch2.convertToResponse(null).isStaleConfig());
sch1.updateActualConfigs(actual);
- // previous value from cache
- Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
- //reset restartRequired flag + invalidating isStale cache
- // after start/restart command execution completed
- sch1.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS/hdfs-site updated host to changed property
Assert.assertFalse(sch1.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch2.convertToResponse(null).isStaleConfig());
@@ -906,7 +872,6 @@ public class ServiceComponentHostTest {
put("dfs_namenode_name_dir", "/foo3"); // HDFS only
put("mapred_log_dir_prefix", "/foo2"); // MR2 only
}}, new HashMap<String, Map<String,String>>());
- waitToStaleConfigsCacheClear();
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
@@ -918,7 +883,6 @@ public class ServiceComponentHostTest {
put("a", "b");
put("fs.trash.interval", "360"); // HDFS only
}}, new HashMap<String, Map<String,String>>());
- waitToStaleConfigsCacheClear();
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
@@ -953,17 +917,10 @@ public class ServiceComponentHostTest {
tags.put(id.toString(), "version2");
actual.put("core-site", tags);
sch3.updateActualConfigs(actual);
- // previous value from cache
- Assert.assertTrue(sch3.convertToResponse(null).isStaleConfig());
- //reset restartRequired flag + invalidating isStale cache
- // after start/restart command execution completed
- sch3.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
Assert.assertFalse(sch3.convertToResponse(null).isStaleConfig());
cluster.deleteConfigGroup(id);
- waitToStaleConfigsCacheClear();
Assert.assertNull(cluster.getConfigGroups().get(id));
sch3.updateActualConfigs(actual);
@@ -971,12 +928,6 @@ public class ServiceComponentHostTest {
tags.remove(id.toString());
sch3.updateActualConfigs(actual);
- // previous value from cache
- Assert.assertTrue(sch3.convertToResponse(null).isStaleConfig());
- //reset restartRequired flag + invalidating isStale cache
- // after start/restart command execution completed
- sch3.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
Assert.assertFalse(sch3.convertToResponse(null).isStaleConfig());
}
@@ -1042,7 +993,6 @@ public class ServiceComponentHostTest {
put("a", "true");
}});
}});
- waitToStaleConfigsCacheClear();
// HDP-x/HDFS does not define type 'foo', so changes do not count to stale
Assert.assertFalse(sch1.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch2.convertToResponse(null).isStaleConfig());
@@ -1052,7 +1002,6 @@ public class ServiceComponentHostTest {
put("mapred-site", new HashMap<String,String>() {{ put("tag", "version1"); }});
}};
sch3.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
sch3.updateActualConfigs(actual);
Assert.assertFalse(sch3.convertToResponse(null).isStaleConfig());
@@ -1067,7 +1016,6 @@ public class ServiceComponentHostTest {
sch1.setRestartRequired(false);
sch2.setRestartRequired(false);
sch3.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch3.convertToResponse(null).isStaleConfig());
@@ -1083,7 +1031,6 @@ public class ServiceComponentHostTest {
sch1.setRestartRequired(false);
sch2.setRestartRequired(false);
sch3.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch3.convertToResponse(null).isStaleConfig());
@@ -1096,7 +1043,6 @@ public class ServiceComponentHostTest {
sch1.setRestartRequired(false);
sch2.setRestartRequired(false);
sch3.setRestartRequired(false);
- waitToStaleConfigsCacheClear();
Assert.assertTrue(sch1.convertToResponse(null).isStaleConfig());
Assert.assertTrue(sch2.convertToResponse(null).isStaleConfig());
Assert.assertFalse(sch3.convertToResponse(null).isStaleConfig());
@@ -1212,20 +1158,4 @@ public class ServiceComponentHostTest {
}
}
}
-
- /*
- Stale configs cache invalidating in separate thread, so sometimes it can not be cleared in time before check.
- So it is needed to wait until thread with cache invalidating complete his work.
- */
- private void waitToStaleConfigsCacheClear() throws NoSuchFieldException, InterruptedException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
- Field f = ConfigHelper.class.getDeclaredField("cacheInvalidationExecutor");
- f.setAccessible(true);
- ExecutorService configHelperExecutor = (ExecutorService) f.get(configHelper);
- configHelperExecutor.shutdown();
- configHelperExecutor.awaitTermination(10, TimeUnit.SECONDS);
-
- Method m = ConfigHelper.class.getDeclaredMethod("createCacheInvalidationExecutor");
- m.setAccessible(true);
- f.set(configHelper, m.invoke(configHelper));
- }
}