This is an automated email from the ASF dual-hosted git repository.
JiaLiangC pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/ambari.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2f794b18ad AMBARI-26614: Config and config-group changes are slow to
propagate to agents on large clusters (#4135)
2f794b18ad is described below
commit 2f794b18ad493f260ad38fff2f5f4651b2cfb16f
Author: Yubi Lee <[email protected]>
AuthorDate: Wed Jun 10 11:13:01 2026 +0900
AMBARI-26614: Config and config-group changes are slow to propagate to
agents on large clusters (#4135)
Reduce the work done on each config / config-group change and fix a config
staleness issue uncovered while doing so.
- Scope the agent-config push: when a config type changes, recompute/push
only to
hosts whose installed components depend on (or whose service owns) that
type
(ConfigHelper.getHostsAffectedByConfigTypes +
AmbariManagementControllerImpl).
Types not owned by any installed service (e.g. cluster-env) still fan out
to all
hosts. Host-set scoping only — each host's config payload content is
unchanged.
- Pre-resolve the changed cluster's desired configs on the request thread
before the
parallel per-host recompute. Resolving them lazily inside the
ForkJoinPool worker
(cachedClustersDesiredConfigs.computeIfAbsent(clusterId, id ->
cl.getDesiredConfigs(false)))
runs outside the request's transaction/persistence context, so it reads
the
pre-change committed snapshot and pushes the previous config value to
agents — a
deterministic off-by-one that leaves a host one change behind until an
ambari-server/agent restart. Seeding the cache on the request thread
(read-your-writes)
keeps the parallel recompute while making the workers never call
getDesiredConfigs.
- Scope config-group changes to the group's member hosts instead of
recomputing
every cluster host (ConfigGroupResourceProvider +
updateAgentConfigs(Cluster, List<Long>)),
with hostId-based group lookup. Covered by
ConfigGroupResourceProviderTest.
- Avoid redundant work per change: cache each cluster's desired configs
once per batch,
drop unnecessary agent-payload sorting (ClusterConfigs) and orphaned
unescape logic.
---
.../server/agent/stomp/AgentConfigsHolder.java | 62 ++++-
.../server/agent/stomp/AgentHostDataHolder.java | 22 ++
.../server/agent/stomp/AlertDefinitionsHolder.java | 7 +
.../server/agent/stomp/HostLevelParamsHolder.java | 7 +
.../server/agent/stomp/dto/ClusterConfigs.java | 8 +
.../controller/AmbariManagementControllerImpl.java | 9 +-
.../internal/ConfigGroupResourceProvider.java | 32 ++-
.../org/apache/ambari/server/state/Cluster.java | 8 +
.../apache/ambari/server/state/ConfigHelper.java | 265 ++++++++++++++-------
.../ambari/server/state/cluster/ClusterImpl.java | 17 ++
.../ambari/server/state/cluster/ClustersImpl.java | 31 ++-
.../apache/ambari/server/state/host/HostImpl.java | 2 +-
.../topology/STOMPComponentsDeleteHandler.java | 8 +-
.../internal/ConfigGroupResourceProviderTest.java | 261 ++++++++++++++++++++
.../ambari/server/state/ConfigHelperTest.java | 41 ++++
15 files changed, 669 insertions(+), 111 deletions(-)
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentConfigsHolder.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentConfigsHolder.java
index 22eebe909e..5c04557751 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentConfigsHolder.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentConfigsHolder.java
@@ -19,6 +19,9 @@ package org.apache.ambari.server.agent.stomp;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import org.apache.ambari.server.AmbariException;
@@ -27,6 +30,7 @@ import
org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.security.encryption.Encryptor;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ConfigHelper;
+import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.utils.ThreadPools;
import org.apache.commons.collections4.CollectionUtils;
@@ -64,10 +68,21 @@ public class AgentConfigsHolder extends
AgentHostDataHolder<AgentConfigsUpdateEv
return configHelper.getHostActualConfigs(hostId);
}
+ @Override
+ public AgentConfigsUpdateEvent getCurrentData(Long hostId,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs)
throws AmbariException {
+ return configHelper.getHostActualConfigs(hostId,
cachedClustersDesiredConfigs);
+ }
+
public AgentConfigsUpdateEvent getCurrentDataExcludeCluster(Long hostId,
Long clusterId) throws AmbariException {
return configHelper.getHostActualConfigsExcludeCluster(hostId, clusterId);
}
+ public AgentConfigsUpdateEvent getCurrentDataExcludeCluster(Long hostId,
Long clusterId,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs)
throws AmbariException {
+ return configHelper.getHostActualConfigsExcludeCluster(hostId, clusterId,
cachedClustersDesiredConfigs);
+ }
+
@Override
protected AgentConfigsUpdateEvent handleUpdate(AgentConfigsUpdateEvent
current, AgentConfigsUpdateEvent update) {
return update;
@@ -84,9 +99,50 @@ public class AgentConfigsHolder extends
AgentHostDataHolder<AgentConfigsUpdateEv
}
}
- for (Long hostId : hostIds) {
- AgentConfigsUpdateEvent agentConfigsUpdateEvent =
configHelper.getHostActualConfigs(hostId);
- updateData(agentConfigsUpdateEvent);
+ final List<Long> targetHostIds = hostIds;
+ // Resolve each host's full (all-cluster) config so a host belonging to
more than one cluster does
+ // not lose the other clusters' configs when its cached event is replaced.
The shared cache
+ // (concurrent for the parallel path) resolves each cluster's desired
configs at most once.
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs = new
ConcurrentHashMap<>();
+ // IMPORTANT - DO NOT MOVE THIS READ INTO THE PARALLEL BLOCK BELOW.
+ // Pre-resolve the changed cluster's desired configs here, on the calling
(request) thread, while
+ // that thread's write transaction is still open. The per-host recompute
below runs on ForkJoinPool
+ // workers, and ConfigHelper.getHostActualConfigsExcludeCluster resolves a
cluster's desired configs
+ // lazily via: cachedClustersDesiredConfigs.computeIfAbsent(clusterId, id
-> cl.getDesiredConfigs(false))
+ // A pool worker is a DIFFERENT thread with no inherited persistence
context/transaction (the
+ // EntityManager/transaction is bound to the writing thread), so that lazy
read sees only the last
+ // COMMITTED snapshot - never this request's still-uncommitted change -
and pushes the PREVIOUS
+ // config value to agents. Because .get() keeps this transaction waiting
for the workers, the worker
+ // read is always pre-commit, producing a deterministic off-by-one: each
change lands on agents one
+ // change late, and the host stays stale until an ambari-server/agent
restart re-resolves it.
+ // Seeding the cache here (read-your-writes on the request thread => fresh
value) makes computeIfAbsent
+ // a no-op for this cluster, so the workers never call getDesiredConfigs
themselves.
+ cachedClustersDesiredConfigs.put(clusterId,
clusters.get().getCluster(clusterId).getDesiredConfigs(false));
+ // Process every host so successful updates are still delivered even if
some fail,
+ // but collect failures and rethrow afterwards so the operator is notified.
+ Map<Long, AmbariException> failures = new ConcurrentHashMap<>();
+ try {
+ // run on the shared default ForkJoinPool so per-host recomputation is
parallelized
+ // without leaking a pool; .get() waits for all hosts to be processed
before returning.
+ threadPools.getDefaultForkJoinPool().submit(() ->
+ targetHostIds.parallelStream().forEach(hostId -> {
+ try {
+ updateData(configHelper.getHostActualConfigs(hostId,
cachedClustersDesiredConfigs));
+ } catch (AmbariException e) {
+ LOG.error("Agent configs update was failed for host {}", hostId,
e);
+ failures.put(hostId, e);
+ }
+ })
+ ).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AmbariException("Agent configs update was interrupted", e);
+ } catch (ExecutionException e) {
+ throw new AmbariException("Agent configs update was failed", e);
+ }
+ if (!failures.isEmpty()) {
+ throw new AmbariException(String.format("Agent configs update failed for
%d of %d host(s): %s",
+ failures.size(), targetHostIds.size(), failures.keySet()),
failures.values().iterator().next());
}
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentHostDataHolder.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentHostDataHolder.java
index 003a2b7945..6ed4b6d959 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentHostDataHolder.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentHostDataHolder.java
@@ -30,6 +30,7 @@ import org.apache.ambari.server.agent.stomp.dto.Hashable;
import org.apache.ambari.server.events.STOMPEvent;
import org.apache.ambari.server.events.STOMPHostEvent;
import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
+import org.apache.ambari.server.state.DesiredConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,6 +47,7 @@ public abstract class AgentHostDataHolder<T extends
STOMPHostEvent & Hashable> e
private final ConcurrentHashMap<Long, T> data = new ConcurrentHashMap<>();
protected abstract T getCurrentData(Long hostId) throws AmbariException;
+ protected abstract T getCurrentData(Long hostId, Map<Long, Map<String,
DesiredConfig>> cachedClustersDesiredConfigs) throws AmbariException;
protected abstract T handleUpdate(T current, T update) throws
AmbariException;
public T getUpdateIfChanged(String agentHash, Long hostId) throws
AmbariException {
@@ -71,6 +73,26 @@ public abstract class AgentHostDataHolder<T extends
STOMPHostEvent & Hashable> e
return hostData;
}
+ public T initializeDataIfNeeded(Long hostId, boolean regenerateHash,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs)
throws AmbariRuntimeException {
+ return data.computeIfAbsent(hostId, id -> initializeData(hostId,
regenerateHash, cachedClustersDesiredConfigs));
+ }
+
+ private T initializeData(Long hostId, boolean regenerateHash,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs) {
+ T hostData;
+ try {
+ hostData = getCurrentData(hostId, cachedClustersDesiredConfigs);
+ } catch (AmbariException e) {
+ LOG.error("Error during retrieving initial value for host: {} and class
{}", hostId, getClass().getName(), e);
+ throw new AmbariRuntimeException("Error during retrieving initial value
for host: " + hostId + " and class: " + getClass().getName(), e);
+ }
+ if (regenerateHash) {
+ regenerateDataIdentifiers(hostData);
+ }
+ return hostData;
+ }
+
/**
* Apply an incremental update to the data (host-specific), and publish the
* event to listeners.
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AlertDefinitionsHolder.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AlertDefinitionsHolder.java
index 4e1edb1bfa..10fe3f3034 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AlertDefinitionsHolder.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AlertDefinitionsHolder.java
@@ -41,6 +41,7 @@ import
org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
import org.apache.ambari.server.orm.entities.AlertDefinitionEntity;
import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.alert.AlertDefinition;
import org.apache.ambari.server.state.alert.AlertDefinitionFactory;
import org.apache.ambari.server.state.alert.AlertDefinitionHash;
@@ -77,6 +78,12 @@ public class AlertDefinitionsHolder extends
AgentHostDataHolder<AlertDefinitions
eventPublisher.register(this);
}
+ @Override
+ protected AlertDefinitionsAgentUpdateEvent getCurrentData(Long hostId,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs)
throws AmbariException {
+ return getCurrentData(hostId);
+ }
+
@Override
protected AlertDefinitionsAgentUpdateEvent getCurrentData(Long hostId)
throws AmbariException {
Map<Long, AlertCluster> result = new TreeMap<>();
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolder.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolder.java
index 5c0efd5b42..7674233830 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolder.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolder.java
@@ -34,6 +34,7 @@ import
org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.state.BlueprintProvisioningState;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.commons.collections4.MapUtils;
@@ -60,6 +61,12 @@ public class HostLevelParamsHolder extends
AgentHostDataHolder<HostLevelParamsUp
ambariEventPublisher.register(this);
}
+ @Override
+ public HostLevelParamsUpdateEvent getCurrentData(Long hostId,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs)
throws AmbariException {
+ return getCurrentData(hostId);
+ }
+
@Override
public HostLevelParamsUpdateEvent getCurrentData(Long hostId) throws
AmbariException {
return getCurrentDataExcludeCluster(hostId, null);
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ClusterConfigs.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ClusterConfigs.java
index 296fb2f0a3..c252060c2a 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ClusterConfigs.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ClusterConfigs.java
@@ -22,6 +22,14 @@ import java.util.SortedMap;
import com.fasterxml.jackson.annotation.JsonInclude;
+/**
+ * Configs sent to an agent for a single cluster.
+ *
+ * The configurations are intentionally kept in {@link SortedMap}s so the JSON
serialization
+ * has a stable, deterministic key order. The per-host config hash is computed
from this JSON
+ * (see AgentDataHolder#getHash); an unstable key order would make the hash
change for unchanged
+ * configs and trigger spurious config updates (and stale-config / restart
prompts) on every agent.
+ */
@JsonInclude(JsonInclude.Include.NON_NULL)
public class ClusterConfigs {
private SortedMap<String, SortedMap<String, String>> configurations;
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
index 90ebcd446b..b859afc8bb 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java
@@ -1852,6 +1852,9 @@ public class AmbariManagementControllerImpl implements
AmbariManagementControlle
new LinkedList<>();
ServiceConfigVersionResponse serviceConfigVersionResponse = null;
boolean nonServiceConfigsChanged = false;
+ // Config types actually changed in this request; used to scope the
agent-config push to the
+ // hosts whose components depend on them (empty => fall back to updating
all hosts).
+ Set<String> changedConfigTypes = new HashSet<>();
if (desiredConfigs != null && request.getServiceConfigVersionRequest() !=
null) {
String msg = "Unable to set desired configs and rollback at same time,
request = " + request;
@@ -1990,6 +1993,7 @@ public class AmbariManagementControllerImpl implements
AmbariManagementControlle
for (Config config : configs) {
Config existingConfig =
cluster.getDesiredConfigByType(config.getType());
existingConfigTypeToConfig.put(config.getType(), existingConfig);
+ changedConfigTypes.add(config.getType());
}
String authName = getAuthName();
@@ -2165,7 +2169,10 @@ public class AmbariManagementControllerImpl implements
AmbariManagementControlle
}
}
if (fireAgentUpdates && (serviceConfigVersionResponse != null ||
nonServiceConfigsChanged)) {
-
configHelper.updateAgentConfigs(Collections.singleton(cluster.getClusterName()));
+ // Scope the agent-config push to hosts whose components depend on the
changed config types.
+ // When changedConfigTypes is empty (e.g. service-config-version revert
path), this falls back
+ // to updating all cluster hosts.
+
configHelper.updateAgentConfigs(Collections.singleton(cluster.getClusterName()),
changedConfigTypes);
}
if (requestStageContainer != null) {
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 8548e45214..c843c2e1bb 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
@@ -552,7 +552,6 @@ public class ConfigGroupResourceProvider extends
ConfigGroupFactory configGroupFactory = getManagementController()
.getConfigGroupFactory();
- Set<String> updatedClusters = new HashSet<>();
for (ConfigGroupRequest request : requests) {
Cluster cluster;
@@ -640,11 +639,14 @@ public class ConfigGroupResourceProvider extends
configGroup.getTag(), configGroup.getDescription(), null, null);
configGroupResponses.add(response);
- updatedClusters.add(cluster.getClusterName());
+ // a new config group only affects its member hosts; skip when it has
none, otherwise an
+ // empty host list would fall back to updating every host in the cluster
(see
+ // AgentConfigsHolder#updateData).
+ if (!hosts.isEmpty()) {
+ m_configHelper.get().updateAgentConfigs(cluster, new
ArrayList<>(hosts.keySet()));
+ }
}
- m_configHelper.get().updateAgentConfigs(updatedClusters);
-
return configGroupResponses;
}
@@ -656,10 +658,10 @@ public class ConfigGroupResourceProvider extends
Clusters clusters = getManagementController().getClusters();
- Set<String> updatedClusters = new HashSet<>();
for (ConfigGroupRequest request : requests) {
Cluster cluster;
+ Set<Long> targetHostIds = new HashSet<>();
try {
cluster = clusters.getCluster(request.getClusterName());
} catch (ClusterNotFoundException e) {
@@ -705,7 +707,12 @@ public class ConfigGroupResourceProvider extends
serviceName = requestServiceName;
}
- int numHosts = (null != configGroup.getHosts()) ?
configGroup.getHosts().size() : 0;
+ int numHosts = 0;
+ Map<Long, Host> prevHosts = configGroup.getHosts();
+ if (null != prevHosts && !prevHosts.isEmpty()) {
+ numHosts = prevHosts.size();
+ targetHostIds.addAll(prevHosts.keySet());
+ }
configLogger.info("(configchange) Updating configuration group host
membership or config value. cluster: '{}', changed by: '{}', " +
"service_name: '{}', config group: '{}', tag: '{}', num hosts in
config group: '{}', note: '{}'",
cluster.getClusterName(), getManagementController().getAuthName(),
@@ -734,7 +741,9 @@ public class ConfigGroupResourceProvider extends
if (hostEntity == null) {
throw new HostNotFoundException(hostname);
}
- hosts.put(hostEntity.getHostId(), host);
+ Long hostId = hostEntity.getHostId();
+ hosts.put(hostId, host);
+ targetHostIds.add(hostId);
}
}
@@ -765,14 +774,17 @@ public class ConfigGroupResourceProvider extends
versionTags.add(tagsMap);
configGroupResponse.setVersionTags(versionTags);
getManagementController().saveConfigGroupUpdate(request,
configGroupResponse);
- updatedClusters.add(cluster.getClusterName());
} else {
LOG.warn("Could not determine service name for config group {},
service config version not created",
configGroup.getId());
}
+ // only the affected hosts (previous + new members) need updating; skip
when there are none,
+ // otherwise an empty host list would fall back to updating every host
in the cluster (see
+ // AgentConfigsHolder#updateData).
+ if (!targetHostIds.isEmpty()) {
+ m_configHelper.get().updateAgentConfigs(cluster, new
ArrayList<>(targetHostIds));
+ }
}
-
- m_configHelper.get().updateAgentConfigs(updatedClusters);
}
@SuppressWarnings("unchecked")
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
b/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
index 799e1dabb7..573570ad2c 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java
@@ -529,6 +529,14 @@ public interface Cluster {
Map<Long, ConfigGroup> getConfigGroupsByHostname(String hostname)
throws AmbariException;
+ /**
+ * Find all config groups associated with the given hostId
+ * @param hostId
+ * @return Map of config group id to config group
+ */
+ Map<Long, ConfigGroup> getConfigGroupsByHostId(Long hostId)
+ throws AmbariException;
+
/**
* Find config group by config group id
* @param configId id of config group to return
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 71efb4edb6..5bf31a83de 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
@@ -62,7 +62,6 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
-import org.apache.commons.lang3.StringEscapeUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1526,7 +1525,7 @@ public class ConfigHelper {
HostConfig hc = actual.get(type);
Map<String, String> actualTags = buildTags(hc);
- if (!isTagChanged(tags, actualTags,
hasGroupSpecificConfigsForType(cluster, sch.getHostName(), type))) {
+ if (!isTagChanged(tags, actualTags,
hasGroupSpecificConfigsForType(cluster, sch.getHost().getHostId(), type))) {
staleEntry = false;
} else {
staleEntry = (serviceInfo.hasConfigDependency(type) ||
componentInfo.hasConfigType(type));
@@ -1576,73 +1575,175 @@ public class ConfigHelper {
clustersInUse.add(cluster);
}
- // get all current and previous host configs
- Map<Long, AgentConfigsUpdateEvent> currentConfigEvents = new HashMap<>();
- Map<Long, AgentConfigsUpdateEvent> previousConfigEvents = new HashMap<>();
for (Cluster cluster : clustersInUse) {
+ Set<Long> hostIds = new HashSet<>();
for (Host host : cluster.getHosts()) {
- Long hostId = host.getHostId();
- if (!currentConfigEvents.containsKey(hostId)) {
- currentConfigEvents.put(host.getHostId(),
m_agentConfigsHolder.get().getCurrentData(hostId));
+ hostIds.add(host.getHostId());
+ }
+ updateAgentConfigs(cluster, new ArrayList<>(hostIds));
+ }
+ }
+
+ /**
+ * Scoped variant of {@link #updateAgentConfigs(Set)}: only hosts running a
component that depends
+ * on (or whose owning service defines) one of the changed config types are
recomputed and pushed.
+ * Global/unowned types (e.g. cluster-env) fan out to all hosts. When the
changed types are unknown
+ * (empty) this falls back to the all-hosts behavior of {@link
#updateAgentConfigs(Set)}.
+ *
+ * @param updatedClusters names of clusters with changed configs
+ * @param changedConfigTypes the config types that actually changed (may be
empty/unknown)
+ * @throws AmbariException
+ */
+ public void updateAgentConfigs(Set<String> updatedClusters, Set<String>
changedConfigTypes) throws AmbariException {
+ if (changedConfigTypes == null || changedConfigTypes.isEmpty()) {
+ // Unknown which types changed: preserve the original all-hosts behavior.
+ updateAgentConfigs(updatedClusters);
+ return;
+ }
+ for (String clusterName : updatedClusters) {
+ Cluster cluster = clusters.getCluster(clusterName);
+ Set<Long> hostIds = getHostsAffectedByConfigTypes(cluster,
changedConfigTypes);
+ if (!hostIds.isEmpty()) {
+ updateAgentConfigs(cluster, new ArrayList<>(hostIds));
+ }
+ }
+ }
+
+ /**
+ * Returns the ids of hosts that run at least one component whose service or
component declares a
+ * dependency on, or whose service owns, any of the given changed config
types. If any changed type
+ * is not owned by a service (e.g. a global type such as cluster-env), every
cluster host is
+ * returned. The resulting set is a strict superset of the hosts the
stale-config logic would mark
+ * for the same types (see {@link #calculateIsStaleConfigs}), so scoping
config updates to it never
+ * skips a host that should be updated.
+ *
+ * @param cluster the cluster
+ * @param changedConfigTypes the changed config types
+ * @return ids of the affected hosts (empty when an owned type has no
installed components)
+ * @throws AmbariException
+ */
+ public Set<Long> getHostsAffectedByConfigTypes(Cluster cluster, Set<String>
changedConfigTypes) throws AmbariException {
+ Set<Long> affectedHostIds = new HashSet<>();
+ if (changedConfigTypes == null || changedConfigTypes.isEmpty()) {
+ return affectedHostIds;
+ }
+
+ // A type not owned by any service is treated as global (e.g.
cluster-env): it affects every host.
+ for (String configType : changedConfigTypes) {
+ if (cluster.getServiceByConfigType(configType) == null) {
+ for (Host host : cluster.getHosts()) {
+ affectedHostIds.add(host.getHostId());
+ }
+ return affectedHostIds;
+ }
+ }
+
+ // Memoize ServiceInfo lookups within this call (avoids O(hosts x
components) metainfo lookups).
+ Map<String, ServiceInfo> serviceInfoCache = new HashMap<>();
+
+ for (Host host : cluster.getHosts()) {
+ for (ServiceComponentHost sch :
cluster.getServiceComponentHosts(host.getHostName())) {
+ StackId stackId = sch.getServiceComponent().getDesiredStackId();
+ String serviceKey = stackId.getStackName() + "/" +
stackId.getStackVersion() + "/" + sch.getServiceName();
+ ServiceInfo serviceInfo = serviceInfoCache.get(serviceKey);
+ if (serviceInfo == null) {
+ serviceInfo = ambariMetaInfo.getService(stackId.getStackName(),
stackId.getStackVersion(), sch.getServiceName());
+ serviceInfoCache.put(serviceKey, serviceInfo);
+ }
+ ComponentInfo componentInfo =
serviceInfo.getComponentByName(sch.getServiceComponentName());
+
+ for (String configType : changedConfigTypes) {
+ // Same relevance the stale-config path uses (hasConfigDependency ||
hasConfigType), plus
+ // service ownership (hasConfigType on the service) to be safe
against incomplete stack
+ // <configuration-dependencies> declarations.
+ if (serviceInfo.hasConfigDependency(configType)
+ || serviceInfo.hasConfigType(configType)
+ || (componentInfo != null &&
componentInfo.hasConfigType(configType))) {
+ affectedHostIds.add(host.getHostId());
+ break;
+ }
}
- if (!previousConfigEvents.containsKey(host.getHostId())) {
- previousConfigEvents.put(host.getHostId(),
- m_agentConfigsHolder.get().initializeDataIfNeeded(hostId, true));
+ if (affectedHostIds.contains(host.getHostId())) {
+ break;
}
}
}
+ return affectedHostIds;
+ }
- for (Cluster cluster : clustersInUse) {
- Map<Long, Map<String, Collection<String>>> changedConfigs = new
HashMap<>();
- for (Host host : cluster.getHosts()) {
- AgentConfigsUpdateEvent currentConfigData =
currentConfigEvents.get(host.getHostId());
- AgentConfigsUpdateEvent previousConfigsData =
previousConfigEvents.get(host.getHostId());
-
- SortedMap<String, SortedMap<String, String>> currentConfigs =
-
currentConfigData.getClustersConfigs().get(Long.toString(cluster.getClusterId())).getConfigurations();
- SortedMap<String, SortedMap<String, String>> previousConfigs =
-
previousConfigsData.getClustersConfigs().get(Long.toString(cluster.getClusterId())).getConfigurations();
-
- Map<String, Collection<String>> changedConfigsHost = new HashMap<>();
- for (String currentConfigType : currentConfigs.keySet()) {
- if (previousConfigs.containsKey(currentConfigType)) {
- Set<String> changedKeys = new HashSet<>();
- Map<String, String> currentTypedConfigs =
currentConfigs.get(currentConfigType);
- Map<String, String> previousTypedConfigs =
previousConfigs.get(currentConfigType);
-
- for (String currentKey : currentTypedConfigs.keySet()) {
- if (!previousTypedConfigs.containsKey(currentKey)
- ||
!currentTypedConfigs.get(currentKey).equals(previousTypedConfigs.get(currentKey)))
{
- changedKeys.add(currentKey);
- }
- }
- for (String previousKey : previousTypedConfigs.keySet()) {
- if (!currentTypedConfigs.containsKey(previousKey)) {
- changedKeys.add(previousKey);
- }
- }
+ /**
+ * Checks populated services for staled configs and updates agent configs
for the given hosts only.
+ * Method retrieves actual agent configs and compares them with just
generated to identify stale configs.
+ * Then config updates are sent to agents.
+ * @param cluster cluster with changed configs
+ * @param hostIds ids of hosts with changed configs
+ * @throws AmbariException
+ */
+ public void updateAgentConfigs(Cluster cluster, List<Long> hostIds) throws
AmbariException {
+
+ // get all current and previous host configs
+ Map<Long, AgentConfigsUpdateEvent> currentConfigEvents = new HashMap<>();
+ Map<Long, AgentConfigsUpdateEvent> previousConfigEvents = new HashMap<>();
+
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs = new
HashMap<>();
+ for (Long hostId : hostIds) {
+ if (!currentConfigEvents.containsKey(hostId)) {
+ currentConfigEvents.put(hostId,
m_agentConfigsHolder.get().getCurrentData(hostId,
cachedClustersDesiredConfigs));
+ }
+ if (!previousConfigEvents.containsKey(hostId)) {
+ previousConfigEvents.put(hostId,
m_agentConfigsHolder.get().initializeDataIfNeeded(hostId, true));
+ }
+ }
+
+ Map<Long, Map<String, Collection<String>>> changedConfigs = new
HashMap<>();
+ for (Long hostId : hostIds) {
+ AgentConfigsUpdateEvent currentConfigData =
currentConfigEvents.get(hostId);
+ AgentConfigsUpdateEvent previousConfigsData =
previousConfigEvents.get(hostId);
+
+ SortedMap<String, SortedMap<String, String>> currentConfigs =
+
currentConfigData.getClustersConfigs().get(Long.toString(cluster.getClusterId())).getConfigurations();
+ SortedMap<String, SortedMap<String, String>> previousConfigs =
+
previousConfigsData.getClustersConfigs().get(Long.toString(cluster.getClusterId())).getConfigurations();
- if (!changedKeys.isEmpty()) {
- changedConfigsHost.put(currentConfigType, changedKeys);
+ Map<String, Collection<String>> changedConfigsHost = new HashMap<>();
+ for (String currentConfigType : currentConfigs.keySet()) {
+ if (previousConfigs.containsKey(currentConfigType)) {
+ Set<String> changedKeys = new HashSet<>();
+ Map<String, String> currentTypedConfigs =
currentConfigs.get(currentConfigType);
+ Map<String, String> previousTypedConfigs =
previousConfigs.get(currentConfigType);
+
+ for (String currentKey : currentTypedConfigs.keySet()) {
+ if (!previousTypedConfigs.containsKey(currentKey)
+ ||
!currentTypedConfigs.get(currentKey).equals(previousTypedConfigs.get(currentKey)))
{
+ changedKeys.add(currentKey);
}
- } else {
- changedConfigsHost.put(currentConfigType,
currentConfigs.get(currentConfigType).keySet());
}
- }
- for (String previousConfigType : previousConfigs.keySet()) {
- if (!currentConfigs.containsKey(previousConfigType)) {
- changedConfigsHost.put(previousConfigType,
previousConfigs.get(previousConfigType).keySet());
+ for (String previousKey : previousTypedConfigs.keySet()) {
+ if (!currentTypedConfigs.containsKey(previousKey)) {
+ changedKeys.add(previousKey);
+ }
+ }
+
+ if (!changedKeys.isEmpty()) {
+ changedConfigsHost.put(currentConfigType, changedKeys);
}
+ } else {
+ changedConfigsHost.put(currentConfigType,
currentConfigs.get(currentConfigType).keySet());
}
- changedConfigs.put(host.getHostId(), changedConfigsHost);
}
- for (String serviceName : cluster.getServices().keySet()) {
- checkStaleConfigsStatusOnConfigsUpdate(cluster.getClusterId(),
serviceName, changedConfigs);
+ for (String previousConfigType : previousConfigs.keySet()) {
+ if (!currentConfigs.containsKey(previousConfigType)) {
+ changedConfigsHost.put(previousConfigType,
previousConfigs.get(previousConfigType).keySet());
+ }
}
-
-
m_metadataHolder.get().updateData(m_ambariManagementController.get().getClusterMetadataOnConfigsUpdate(cluster));
- m_agentConfigsHolder.get().updateData(cluster.getClusterId(), null);
+ changedConfigs.put(hostId, changedConfigsHost);
}
+ for (String serviceName : cluster.getServices().keySet()) {
+ checkStaleConfigsStatusOnConfigsUpdate(cluster.getClusterId(),
serviceName, changedConfigs);
+ }
+
+
m_metadataHolder.get().updateData(m_ambariManagementController.get().getClusterMetadataOnConfigsUpdate(cluster));
+ m_agentConfigsHolder.get().updateData(cluster.getClusterId(), hostIds);
}
/**
@@ -1910,13 +2011,13 @@ public class ConfigHelper {
* Determines if the hostname has group specific configs for the type
specified
*
* @param cluster
- * @param hostname of the host to look for
+ * @param hostId of the host to look for
* @param type the type to look for (e.g. flume-conf)
* @return <code>true</code> if the hostname has group specific
configuration for the type
*/
- private boolean hasGroupSpecificConfigsForType(Cluster cluster, String
hostname, String type) {
+ private boolean hasGroupSpecificConfigsForType(Cluster cluster, Long hostId,
String type) {
try {
- Map<Long, ConfigGroup> configGroups =
cluster.getConfigGroupsByHostname(hostname);
+ Map<Long, ConfigGroup> configGroups =
cluster.getConfigGroupsByHostId(hostId);
if (configGroups != null && !configGroups.isEmpty()) {
for (ConfigGroup configGroup : configGroups.values()) {
Config config = configGroup.getConfigurations().get(type);
@@ -2098,7 +2199,17 @@ public class ConfigHelper {
return getHostActualConfigsExcludeCluster(hostId, null);
}
+ public AgentConfigsUpdateEvent getHostActualConfigs(Long hostId,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs)
throws AmbariException {
+ return getHostActualConfigsExcludeCluster(hostId, null,
cachedClustersDesiredConfigs);
+ }
+
public AgentConfigsUpdateEvent getHostActualConfigsExcludeCluster(Long
hostId, Long clusterId) throws AmbariException {
+ return getHostActualConfigsExcludeCluster(hostId, clusterId, new
HashMap<>());
+ }
+
+ public AgentConfigsUpdateEvent getHostActualConfigsExcludeCluster(Long
hostId, Long clusterId,
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs)
throws AmbariException {
TreeMap<String, ClusterConfigs> clustersConfigs = new TreeMap<>();
Host host = clusters.getHostById(hostId);
@@ -2108,7 +2219,10 @@ public class ConfigHelper {
}
Map<String, Map<String, String>> configurations = new HashMap<>();
Map<String, Map<String, Map<String, String>>> configurationAttributes =
new HashMap<>();
- Map<String, DesiredConfig> clusterDesiredConfigs =
cl.getDesiredConfigs(false);
+ // computeIfAbsent so the desired configs are resolved at most once per
cluster, even when this
+ // cache is shared across threads (the parallel host-init path uses a
ConcurrentHashMap).
+ Map<String, DesiredConfig> clusterDesiredConfigs =
+ cachedClustersDesiredConfigs.computeIfAbsent(cl.getClusterId(), id
-> cl.getDesiredConfigs(false));
Map<String, Map<String, String>> configTags =
getEffectiveDesiredTags(cl, host.getHostName(), clusterDesiredConfigs);
// Logging below creating too much spam and slowing down operations
@@ -2119,49 +2233,26 @@ public class ConfigHelper {
}
getAndMergeHostConfigs(configurations, configTags, cl);
- configurations = unescapeConfigNames(configurations);
getAndMergeHostConfigAttributes(configurationAttributes, configTags, cl);
- configurationAttributes =
unescapeConfigAttributeNames(configurationAttributes);
- SortedMap<String, SortedMap<String, String>> configurationsTreeMap =
sortConfigutations(configurations);
- SortedMap<String, SortedMap<String, SortedMap<String, String>>>
configurationAttributesTreeMap =
- sortConfigurationAttributes(configurationAttributes);
+ // Keep the payload sorted so the per-host config hash (computed from
the JSON serialization)
+ // is stable for unchanged configs; see ClusterConfigs.
clustersConfigs.put(Long.toString(cl.getClusterId()),
- new ClusterConfigs(configurationsTreeMap,
configurationAttributesTreeMap));
+ new ClusterConfigs(sortConfigurations(configurations),
sortConfigurationAttributes(configurationAttributes)));
}
return new AgentConfigsUpdateEvent(hostId, clustersConfigs);
}
- private Map<String, Map<String, String>> unescapeConfigNames(Map<String,
Map<String, String>> configurations) {
- Map<String, Map<String, String>> unescapedConfigs = new HashMap<>();
- for (Entry<String, Map<String, String>> configTypeEntry :
configurations.entrySet()) {
- Map<String, String> unescapedTypeConfigs = new HashMap<>();
- for (Entry<String, String> config :
configTypeEntry.getValue().entrySet()) {
-
unescapedTypeConfigs.put(StringEscapeUtils.unescapeJava(config.getKey()),
config.getValue());
- }
- unescapedConfigs.put(configTypeEntry.getKey(), unescapedTypeConfigs);
- }
-
- return unescapedConfigs;
- }
-
- private Map<String, Map<String, Map<String, String>>>
unescapeConfigAttributeNames(
- Map<String, Map<String, Map<String, String>>> configurationAttributes) {
- Map<String, Map<String, Map<String, String>>> unescapedConfigAttributes =
new HashMap<>();
-
- configurationAttributes.forEach((key, value) ->
unescapedConfigAttributes.put(key, unescapeConfigNames(value)));
- return unescapedConfigAttributes;
- }
-
- public SortedMap<String, SortedMap<String, String>>
sortConfigutations(Map<String, Map<String, String>> configurations) {
+ private SortedMap<String, SortedMap<String, String>>
sortConfigurations(Map<String, Map<String, String>> configurations) {
SortedMap<String, SortedMap<String, String>> configurationsTreeMap = new
TreeMap<>();
configurations.forEach((k, v) -> configurationsTreeMap.put(k, new
TreeMap<>(v)));
return configurationsTreeMap;
}
- public SortedMap<String, SortedMap<String, SortedMap<String, String>>>
sortConfigurationAttributes(
+ private SortedMap<String, SortedMap<String, SortedMap<String, String>>>
sortConfigurationAttributes(
Map<String, Map<String, Map<String, String>>> configurationAttributes) {
+
SortedMap<String, SortedMap<String, SortedMap<String, String>>>
configurationAttributesTreeMap = new TreeMap<>();
configurationAttributes.forEach((k, v) -> {
SortedMap<String, SortedMap<String, String>> c = new TreeMap<>();
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 c4eb82176a..93bc00402c 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
@@ -535,6 +535,23 @@ public class ClusterImpl implements Cluster {
return configGroups;
}
+ @Override
+ public Map<Long, ConfigGroup> getConfigGroupsByHostId(Long hostId) {
+ Map<Long, ConfigGroup> configGroups = new HashMap<>();
+
+ for (Entry<Long, ConfigGroup> groupEntry : clusterConfigGroups.entrySet())
{
+ Long id = groupEntry.getKey();
+ ConfigGroup group = groupEntry.getValue();
+ for (Host host : group.getHosts().values()) {
+ if (Objects.equals(hostId, host.getHostId())) {
+ configGroups.put(id, group);
+ break;
+ }
+ }
+ }
+ return configGroups;
+ }
+
@Override
public ConfigGroup getConfigGroupsById(Long configId) {
return clusterConfigGroups.get(configId);
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClustersImpl.java
b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClustersImpl.java
index 10497e100a..df381eea8b 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClustersImpl.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClustersImpl.java
@@ -30,6 +30,7 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
import jakarta.persistence.RollbackException;
@@ -76,6 +77,7 @@ import
org.apache.ambari.server.security.authorization.ResourceType;
import org.apache.ambari.server.state.AgentVersion;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostHealthStatus;
import org.apache.ambari.server.state.HostHealthStatus.HealthStatus;
@@ -86,6 +88,7 @@ import org.apache.ambari.server.state.configgroup.ConfigGroup;
import org.apache.ambari.server.state.host.HostFactory;
import org.apache.ambari.server.topology.TopologyManager;
import org.apache.ambari.server.utils.RetryHelper;
+import org.apache.ambari.server.utils.ThreadPools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.GrantedAuthority;
@@ -157,6 +160,9 @@ public class ClustersImpl implements Clusters {
@Inject
private Provider<AgentConfigsHolder> m_agentConfigsHolder;
+ @Inject
+ private ThreadPools threadPools;
+
@Inject
private Provider<MetadataHolder> m_metadataHolder;
@@ -318,13 +324,24 @@ public class ClustersImpl implements Clusters {
}
hostClustersMap = hostClustersMapTemp;
clusterHostsMap1 = clusterHostsMap1Temp;
- // init host configs
- for (Long hostId : hostsById.keySet()) {
- try {
- m_agentConfigsHolder.get().initializeDataIfNeeded(hostId, true);
- } catch (AmbariRuntimeException e) {
- LOG.error("Agent configs initialization was failed", e);
- }
+ // init host configs in parallel on the shared default ForkJoinPool; the
per-batch cache
+ // must be concurrent since it is populated from multiple threads.
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs = new
ConcurrentHashMap<>();
+ try {
+ threadPools.getDefaultForkJoinPool().submit(() ->
+ hostsById.keySet().parallelStream().forEach(hostId -> {
+ try {
+ m_agentConfigsHolder.get().initializeDataIfNeeded(hostId, true,
cachedClustersDesiredConfigs);
+ } catch (AmbariRuntimeException e) {
+ LOG.error("Agent configs initialization was failed", e);
+ }
+ })
+ ).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.error("Agent configs initialization was failed", e);
+ } catch (ExecutionException e) {
+ LOG.error("Agent configs initialization was failed", e);
}
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/state/host/HostImpl.java
b/ambari-server/src/main/java/org/apache/ambari/server/state/host/HostImpl.java
index 8702001c1a..7fd1a25fc2 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/state/host/HostImpl.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/state/host/HostImpl.java
@@ -1125,7 +1125,7 @@ public class HostImpl implements Host {
}
));
- Map<Long, ConfigGroup> configGroups = (cluster == null) ? new HashMap<>()
: cluster.getConfigGroupsByHostname(getHostName());
+ Map<Long, ConfigGroup> configGroups = (cluster == null) ? new HashMap<>()
: cluster.getConfigGroupsByHostId(getHostId());
if (configGroups == null || configGroups.isEmpty()) {
return hostConfigMap;
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/topology/STOMPComponentsDeleteHandler.java
b/ambari-server/src/main/java/org/apache/ambari/server/topology/STOMPComponentsDeleteHandler.java
index 79dd94836e..7bd97cc501 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/topology/STOMPComponentsDeleteHandler.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/topology/STOMPComponentsDeleteHandler.java
@@ -18,7 +18,9 @@
package org.apache.ambari.server.topology;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Collectors;
@@ -35,6 +37,7 @@ import
org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMet
import org.apache.ambari.server.events.TopologyUpdateEvent;
import org.apache.ambari.server.events.UpdateEventType;
import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.DesiredConfig;
import com.google.inject.Inject;
import com.google.inject.Provider;
@@ -95,13 +98,14 @@ public class STOMPComponentsDeleteHandler {
private void updateNonTopologyAgentInfo(Set<Long> changedHosts, Long
clusterId) throws AmbariException {
+ Map<Long, Map<String, DesiredConfig>> cachedClustersDesiredConfigs = new
HashMap<>();
for (Long hostId : changedHosts) {
if (clusterId != null) {
alertDefinitionsHolder.get().updateData(alertDefinitionsHolder.get().getDeleteCluster(clusterId,
hostId));
-
agentConfigsHolder.get().updateData(agentConfigsHolder.get().getCurrentDataExcludeCluster(hostId,
clusterId));
+
agentConfigsHolder.get().updateData(agentConfigsHolder.get().getCurrentDataExcludeCluster(hostId,
clusterId, cachedClustersDesiredConfigs));
hostLevelParamsHolder.get().updateData(hostLevelParamsHolder.get().getCurrentDataExcludeCluster(hostId,
clusterId));
} else {
-
agentConfigsHolder.get().updateData(agentConfigsHolder.get().getCurrentData(hostId));
+
agentConfigsHolder.get().updateData(agentConfigsHolder.get().getCurrentData(hostId,
cachedClustersDesiredConfigs));
hostLevelParamsHolder.get().updateData(hostLevelParamsHolder.get().getCurrentData(hostId));
}
}
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 449d55047e..381eac154e 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
@@ -582,6 +582,267 @@ public class ConfigGroupResourceProviderTest {
configGroup, response, configGroupResponse, configHelper, hostDAO,
hostEntity1, hostEntity2, h1, h2);
}
+ @Test
+ public void testUpdateConfigGroupScopesToPreviousAndNewMemberHosts() throws
Exception {
+ // The config group already has members h1 and h2; the update keeps only
h1.
+ // Agent config update must be scoped to the union of previous and new
members (h1, h2) so the
+ // removed host (h2) is reverted to the cluster-default config.
+ Authentication authentication =
TestAuthenticationFactory.createAdministrator();
+
+ AmbariManagementController managementController =
createMock(AmbariManagementController.class);
+ RequestStatusResponse response =
createNiceMock(RequestStatusResponse.class);
+ ConfigHelper configHelper = createMock(ConfigHelper.class);
+ Clusters clusters = createNiceMock(Clusters.class);
+ Cluster cluster = createNiceMock(Cluster.class);
+ Host h1 = createNiceMock(Host.class);
+ Host h2 = createNiceMock(Host.class);
+ HostEntity hostEntity1 = createMock(HostEntity.class);
+
+ final ConfigGroup configGroup = createNiceMock(ConfigGroup.class);
+ ConfigGroupResponse configGroupResponse =
createNiceMock(ConfigGroupResponse.class);
+
+ expect(cluster.isConfigTypeExists("core-site")).andReturn(true).anyTimes();
+ expect(managementController.getClusters()).andReturn(clusters).anyTimes();
+ expect(managementController.getAuthName()).andReturn("admin").anyTimes();
+ expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
+ expect(clusters.getHost("h1")).andReturn(h1);
+ expect(hostDAO.findByName("h1")).andReturn(hostEntity1).anyTimes();
+ expect(hostDAO.findById(1L)).andReturn(hostEntity1).anyTimes();
+ expect(hostEntity1.getHostId()).andReturn(1L).atLeastOnce();
+ expect(h1.getHostId()).andReturn(1L).anyTimes();
+ expect(h2.getHostId()).andReturn(2L).anyTimes();
+
expect(managementController.getConfigGroupUpdateResults((ConfigGroupRequest)
anyObject())).
+ andReturn(new ConfigGroupResponse(1L, "", "", "", "", new HashSet<>(),
new HashSet<>())).atLeastOnce();
+
+ expect(configGroup.getName()).andReturn("test-1").anyTimes();
+ expect(configGroup.getId()).andReturn(25L).anyTimes();
+ expect(configGroup.getTag()).andReturn("tag-1").anyTimes();
+ // existing members of the group before the update
+ Map<Long, Host> existingHosts = new HashMap<>();
+ existingHosts.put(1L, h1);
+ existingHosts.put(2L, h2);
+ expect(configGroup.getHosts()).andReturn(existingHosts).anyTimes();
+
expect(configGroup.convertToResponse()).andReturn(configGroupResponse).anyTimes();
+
expect(configGroupResponse.getClusterName()).andReturn("Cluster100").anyTimes();
+ expect(configGroupResponse.getId()).andReturn(25L).anyTimes();
+
+ expect(cluster.getConfigGroups()).andStubAnswer(new IAnswer<Map<Long,
ConfigGroup>>() {
+ @Override
+ public Map<Long, ConfigGroup> answer() throws Throwable {
+ Map<Long, ConfigGroup> configGroupMap = new HashMap<>();
+ configGroupMap.put(configGroup.getId(), configGroup);
+ return configGroupMap;
+ }
+ });
+
+ // capture the hosts the agent config update is scoped to
+ Capture<List<Long>> hostIdsCapture = newCapture();
+ configHelper.updateAgentConfigs(anyObject(Cluster.class),
capture(hostIdsCapture));
+
+ replay(managementController, clusters, cluster, configGroup, response,
configGroupResponse,
+ configHelper, hostDAO, hostEntity1, h1, h2);
+
+ ConfigGroupResourceProvider provider = (ConfigGroupResourceProvider)
+
AbstractControllerResourceProvider.getResourceProvider(Resource.Type.ConfigGroup,
managementController);
+ Provider<ConfigHelper> configHelperProvider =
createNiceMock(Provider.class);
+ expect(configHelperProvider.get()).andReturn(configHelper).anyTimes();
+ replay(configHelperProvider);
+ Field m_configHelper =
ConfigGroupResourceProvider.class.getDeclaredField("m_configHelper");
+ m_configHelper.setAccessible(true);
+ m_configHelper.set(provider, configHelperProvider);
+
+ Map<String, Object> properties = new LinkedHashMap<>();
+
+ Set<Map<String, Object>> hostSet = new HashSet<>();
+ Map<String, Object> host1 = new HashMap<>();
+ host1.put(ConfigGroupResourceProvider.HOST_NAME, "h1");
+ hostSet.add(host1);
+
+ Set<Map<String, Object>> configSet = new HashSet<>();
+ Map<String, String> configMap = new HashMap<>();
+ Map<String, Object> configs = new HashMap<>();
+ configs.put("type", "core-site");
+ configs.put("tag", "version100");
+ configMap.put("key1", "value1");
+ configs.put("properties", configMap);
+ configSet.add(configs);
+
+ properties.put(ConfigGroupResourceProvider.CLUSTER_NAME, "Cluster100");
+ properties.put(ConfigGroupResourceProvider.GROUP_NAME, "test-1");
+ properties.put(ConfigGroupResourceProvider.TAG, "tag-1");
+ properties.put(ConfigGroupResourceProvider.HOSTS, hostSet);
+ properties.put(ConfigGroupResourceProvider.DESIRED_CONFIGS, configSet);
+
+ Map<String, String> mapRequestProps = new HashMap<>();
+ mapRequestProps.put("context", "Called from a test");
+
+ Request request = PropertyHelper.getUpdateRequest(properties,
mapRequestProps);
+
+ Predicate predicate = new PredicateBuilder().property
+ (ConfigGroupResourceProvider.CLUSTER_NAME).equals("Cluster100").and().
+ property(ConfigGroupResourceProvider.ID).equals(25L).toPredicate();
+
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ provider.updateResources(request, predicate);
+
+ verify(configHelper);
+ List<Long> capturedHostIds = hostIdsCapture.getValue();
+ assertEquals(2, capturedHostIds.size());
+ assertTrue(capturedHostIds.contains(1L));
+ assertTrue(capturedHostIds.contains(2L));
+ }
+
+ @Test
+ public void testCreateConfigGroupWithoutHostsSkipsAgentConfigUpdate() throws
Exception {
+ // A config group created with no member hosts must not trigger an agent
config update;
+ // an empty host list would otherwise fall back to updating every host in
the cluster
+ // (see AgentConfigsHolder#updateData).
+ Authentication authentication =
TestAuthenticationFactory.createAdministrator();
+
+ AmbariManagementController managementController =
createMock(AmbariManagementController.class);
+ Clusters clusters = createNiceMock(Clusters.class);
+ Cluster cluster = createNiceMock(Cluster.class);
+ ConfigGroupFactory configGroupFactory =
createNiceMock(ConfigGroupFactory.class);
+ ConfigGroup configGroup = createNiceMock(ConfigGroup.class);
+ ConfigHelper configHelper = createMock(ConfigHelper.class);
+
+ expect(managementController.getClusters()).andReturn(clusters).anyTimes();
+ expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
+ expect(cluster.getClusterName()).andReturn("Cluster100").anyTimes();
+ expect(cluster.isConfigTypeExists(anyString())).andReturn(true).anyTimes();
+
expect(managementController.getConfigGroupFactory()).andReturn(configGroupFactory).anyTimes();
+ expect(managementController.getAuthName()).andReturn("admin").anyTimes();
+ expect(configGroupFactory.createNew(anyObject(Cluster.class), anyObject(),
anyObject(),
+ anyObject(), anyObject(), anyObject(),
anyObject())).andReturn(configGroup).anyTimes();
+
+ // configHelper.updateAgentConfigs is intentionally NOT expected; the
strict mock fails if it is called.
+ replay(managementController, clusters, cluster, configGroupFactory,
configGroup, configHelper);
+
+ ConfigGroupResourceProvider provider = (ConfigGroupResourceProvider)
+
AbstractControllerResourceProvider.getResourceProvider(Resource.Type.ConfigGroup,
managementController);
+ Provider<ConfigHelper> configHelperProvider =
createNiceMock(Provider.class);
+ expect(configHelperProvider.get()).andReturn(configHelper).anyTimes();
+ replay(configHelperProvider);
+ Field m_configHelper =
ConfigGroupResourceProvider.class.getDeclaredField("m_configHelper");
+ m_configHelper.setAccessible(true);
+ m_configHelper.set(provider, configHelperProvider);
+
+ Set<Map<String, Object>> propertySet = new LinkedHashSet<>();
+ Map<String, Object> properties = new LinkedHashMap<>();
+
+ Set<Map<String, Object>> configSet = new HashSet<>();
+ Map<String, String> configMap = new HashMap<>();
+ Map<String, Object> configs = new HashMap<>();
+ configs.put("type", "core-site");
+ configs.put("tag", "version100");
+ configMap.put("key1", "value1");
+ configs.put("properties", configMap);
+ configSet.add(configs);
+
+ properties.put(ConfigGroupResourceProvider.CLUSTER_NAME, "Cluster100");
+ properties.put(ConfigGroupResourceProvider.GROUP_NAME, "test-1");
+ properties.put(ConfigGroupResourceProvider.TAG, "tag-1");
+ // no HOSTS property: the group has no member hosts
+ properties.put(ConfigGroupResourceProvider.DESIRED_CONFIGS, configSet);
+ propertySet.add(properties);
+
+ Request request = PropertyHelper.getCreateRequest(propertySet, null);
+
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ provider.createResources(request);
+
+ // strict configHelper mock verifies updateAgentConfigs was never invoked
+ verify(configHelper);
+ }
+
+ @Test
+ public void testUpdateConfigGroupWithoutHostsSkipsAgentConfigUpdate() throws
Exception {
+ // Updating a config group to have no member hosts (and it had none
before) must not trigger an
+ // agent config update; an empty host list would otherwise fall back to
updating every host in
+ // the cluster (see AgentConfigsHolder#updateData).
+ Authentication authentication =
TestAuthenticationFactory.createAdministrator();
+
+ AmbariManagementController managementController =
createMock(AmbariManagementController.class);
+ ConfigHelper configHelper = createMock(ConfigHelper.class);
+ Clusters clusters = createNiceMock(Clusters.class);
+ Cluster cluster = createNiceMock(Cluster.class);
+
+ final ConfigGroup configGroup = createNiceMock(ConfigGroup.class);
+ ConfigGroupResponse configGroupResponse =
createNiceMock(ConfigGroupResponse.class);
+
+ expect(cluster.isConfigTypeExists("core-site")).andReturn(true).anyTimes();
+ expect(managementController.getClusters()).andReturn(clusters).anyTimes();
+ expect(managementController.getAuthName()).andReturn("admin").anyTimes();
+ expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
+
expect(managementController.getConfigGroupUpdateResults((ConfigGroupRequest)
anyObject())).
+ andReturn(new ConfigGroupResponse(1L, "", "", "", "", new HashSet<>(),
new HashSet<>())).anyTimes();
+
+ expect(configGroup.getName()).andReturn("test-1").anyTimes();
+ expect(configGroup.getId()).andReturn(25L).anyTimes();
+ expect(configGroup.getTag()).andReturn("tag-1").anyTimes();
+ // the group has no members before the update, and the update sets none
+ expect(configGroup.getHosts()).andReturn(new HashMap<>()).anyTimes();
+
expect(configGroup.convertToResponse()).andReturn(configGroupResponse).anyTimes();
+
expect(configGroupResponse.getClusterName()).andReturn("Cluster100").anyTimes();
+ expect(configGroupResponse.getId()).andReturn(25L).anyTimes();
+
+ expect(cluster.getConfigGroups()).andStubAnswer(new IAnswer<Map<Long,
ConfigGroup>>() {
+ @Override
+ public Map<Long, ConfigGroup> answer() throws Throwable {
+ Map<Long, ConfigGroup> configGroupMap = new HashMap<>();
+ configGroupMap.put(configGroup.getId(), configGroup);
+ return configGroupMap;
+ }
+ });
+
+ // configHelper.updateAgentConfigs is intentionally NOT expected; the
strict mock fails if it is called.
+ replay(managementController, clusters, cluster, configGroup,
configGroupResponse, configHelper);
+
+ ConfigGroupResourceProvider provider = (ConfigGroupResourceProvider)
+
AbstractControllerResourceProvider.getResourceProvider(Resource.Type.ConfigGroup,
managementController);
+ Provider<ConfigHelper> configHelperProvider =
createNiceMock(Provider.class);
+ expect(configHelperProvider.get()).andReturn(configHelper).anyTimes();
+ replay(configHelperProvider);
+ Field m_configHelper =
ConfigGroupResourceProvider.class.getDeclaredField("m_configHelper");
+ m_configHelper.setAccessible(true);
+ m_configHelper.set(provider, configHelperProvider);
+
+ Map<String, Object> properties = new LinkedHashMap<>();
+
+ Set<Map<String, Object>> configSet = new HashSet<>();
+ Map<String, String> configMap = new HashMap<>();
+ Map<String, Object> configs = new HashMap<>();
+ configs.put("type", "core-site");
+ configs.put("tag", "version100");
+ configMap.put("key1", "value1");
+ configs.put("properties", configMap);
+ configSet.add(configs);
+
+ properties.put(ConfigGroupResourceProvider.CLUSTER_NAME, "Cluster100");
+ properties.put(ConfigGroupResourceProvider.GROUP_NAME, "test-1");
+ properties.put(ConfigGroupResourceProvider.TAG, "tag-1");
+ // no HOSTS property: the update leaves the group with no member hosts
+ properties.put(ConfigGroupResourceProvider.DESIRED_CONFIGS, configSet);
+
+ Map<String, String> mapRequestProps = new HashMap<>();
+ mapRequestProps.put("context", "Called from a test");
+
+ Request request = PropertyHelper.getUpdateRequest(properties,
mapRequestProps);
+
+ Predicate predicate = new PredicateBuilder().property
+ (ConfigGroupResourceProvider.CLUSTER_NAME).equals("Cluster100").and().
+ property(ConfigGroupResourceProvider.ID).equals(25L).toPredicate();
+
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ provider.updateResources(request, predicate);
+
+ // strict configHelper mock verifies updateAgentConfigs was never invoked
+ verify(configHelper);
+ }
+
@Test
public void testGetConfigGroupAsAmbariAdministrator() throws Exception {
testGetConfigGroup(TestAuthenticationFactory.createAdministrator());
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 2955d1e1d7..136811daf6 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
@@ -363,6 +363,45 @@ public class ConfigHelperTest {
Assert.assertEquals(expectedConfig_hiveServer1,
originalConfig_hiveServer1);
}
+ @Test
+ public void testGetHostsAffectedByConfigTypes() throws Exception {
+ Long h1 = clusters.getHost("h1").getHostId();
+ Long h2 = clusters.getHost("h2").getHostId();
+ Long h3 = clusters.getHost("h3").getHostId();
+ clusters.mapHostToCluster("h1", cluster.getClusterName());
+ clusters.mapHostToCluster("h2", cluster.getClusterName());
+ clusters.mapHostToCluster("h3", cluster.getClusterName());
+
+ // Empty changed types -> empty set (the caller falls back to all-hosts).
+ Assert.assertTrue(configHelper.getHostsAffectedByConfigTypes(cluster,
new HashSet<String>()).isEmpty());
+
+ // A config type owned by no installed service (global, e.g.
cluster-env) -> all hosts.
+ Set<String> globalType = new HashSet<>();
+ globalType.add("a-config-type-no-service-owns");
+ Set<Long> expectedAllHosts = new HashSet<>();
+ expectedAllHosts.add(h1);
+ expectedAllHosts.add(h2);
+ expectedAllHosts.add(h3);
+ Assert.assertEquals(expectedAllHosts,
configHelper.getHostsAffectedByConfigTypes(cluster, globalType));
+
+ // cluster-env is a real stack-level config owned by no service -> must
be treated as global.
+ Assert.assertNull("cluster-env must be unowned/global",
cluster.getServiceByConfigType("cluster-env"));
+ Set<String> clusterEnv = new HashSet<>();
+ clusterEnv.add("cluster-env");
+ Assert.assertEquals(expectedAllHosts,
configHelper.getHostsAffectedByConfigTypes(cluster, clusterEnv));
+
+ // hdfs-site is owned by the (installed) HDFS service, but no host runs
an HDFS component yet -> empty.
+ Set<String> hdfsSite = new HashSet<>();
+ hdfsSite.add("hdfs-site");
+ Assert.assertTrue(configHelper.getHostsAffectedByConfigTypes(cluster,
hdfsSite).isEmpty());
+
+ // Install an HDFS component on h1; hdfs-site now affects only h1.
+
cluster.getService("HDFS").addServiceComponent("NAMENODE").addServiceComponentHost("h1");
+ Set<Long> expectedH1Only = new HashSet<>();
+ expectedH1Only.add(h1);
+ Assert.assertEquals(expectedH1Only,
configHelper.getHostsAffectedByConfigTypes(cluster, hdfsSite));
+ }
+
private Map<String, Map<String, String>> createHiveConfig() {
return new HashMap<String, Map<String, String>>() {{
put("hive-site", new HashMap<String, String>() {{
@@ -983,6 +1022,7 @@ public class ConfigHelperTest {
// set up expectations
expect(sch.getActualConfigs()).andReturn(schReturn).times(6);
expect(sch.getHostName()).andReturn("h1").anyTimes();
+ expect(sch.getHost()).andReturn(clusters.getHost("h1")).anyTimes();
expect(sch.getClusterId()).andReturn(cluster.getClusterId()).anyTimes();
expect(sch.getServiceName()).andReturn("FLUME").anyTimes();
expect(sch.getServiceComponentName()).andReturn("FLUME_HANDLER").anyTimes();
@@ -1058,6 +1098,7 @@ public class ConfigHelperTest {
// set up expectations
expect(sch.getActualConfigs()).andReturn(schReturn).anyTimes();
expect(sch.getHostName()).andReturn("h1").anyTimes();
+ expect(sch.getHost()).andReturn(clusters.getHost("h1")).anyTimes();
expect(sch.getClusterId()).andReturn(cluster.getClusterId()).anyTimes();
expect(sch.getServiceName()).andReturn("HDFS").anyTimes();
expect(sch.getServiceComponentName()).andReturn("NAMENODE").anyTimes();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
