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 1e753e06ab AMBARI-26640: Prevent auto-recovery from blocking cluster
start operations (#4187)
1e753e06ab is described below
commit 1e753e06abb2d5219b01a80befe57301381a143f
Author: jialiang <[email protected]>
AuthorDate: Mon Aug 31 08:42:19 2026 +0000
AMBARI-26640: Prevent auto-recovery from blocking cluster start operations
(#4187)
* AMBARI-26640: Add stateful recovery topology model
* AMBARI-26640: Track recovery topology by agent session
* AMBARI-26640: Prevent auto-recovery from blocking cluster start operations
* AMBARI-26640: Handle recovery topology during service removal
* AMBARI-26640: Retrigger CI after Node cache failure
---
.../src/main/python/ambari_agent/ActionQueue.py | 100 ++++++++++++---
.../python/ambari_agent/ComponentStatusExecutor.py | 98 +++++++++++---
.../main/python/ambari_agent/HeartbeatThread.py | 5 +-
.../main/python/ambari_agent/RecoveryManager.py | 86 ++++++++++++-
.../test/python/ambari_agent/TestActionQueue.py | 85 +++++++++++++
.../ambari_agent/TestComponentStatusExecutor.py | 91 +++++++++++++
.../python/ambari_agent/TestRecoveryManager.py | 141 +++++++++++++++++++++
.../server/agent/ComponentStatusAgentReport.java | 26 +++-
.../ambari/server/agent/HeartbeatMonitor.java | 8 ++
.../apache/ambari/server/agent/RecoveryConfig.java | 46 ++++++-
.../server/agent/RecoveryConfigComponent.java | 23 +++-
.../server/agent/RecoveryConfigDependency.java | 113 +++++++++++++++++
.../ambari/server/agent/RecoveryConfigHelper.java | 81 +++++++++++-
.../server/agent/RecoveryTopologyManager.java | 82 ++++++++++++
.../server/agent/stomp/AgentReportsController.java | 12 +-
.../server/agent/stomp/HeartbeatController.java | 12 +-
.../server/agent/stomp/HostLevelParamsHolder.java | 18 ++-
.../agent/stomp/dto/ComponentStatusReports.java | 10 ++
.../ambari/server/metadata/RoleCommandOrder.java | 9 ++
.../ambari/server/metadata/RoleCommandPair.java | 4 +-
.../agent/ComponentStatusAgentReportTest.java | 45 +++++++
.../server/agent/RecoveryTopologyManagerTest.java | 59 +++++++++
.../agent/stomp/HostLevelParamsHolderTest.java | 17 +++
.../configuration/RecoveryConfigHelperTest.java | 51 ++++++++
24 files changed, 1172 insertions(+), 50 deletions(-)
diff --git a/ambari-agent/src/main/python/ambari_agent/ActionQueue.py
b/ambari-agent/src/main/python/ambari_agent/ActionQueue.py
index f0f9f6ef76..9a168fde71 100644
--- a/ambari-agent/src/main/python/ambari_agent/ActionQueue.py
+++ b/ambari-agent/src/main/python/ambari_agent/ActionQueue.py
@@ -80,6 +80,8 @@ class ActionQueue(threading.Thread):
self.parallel_execution = self.config.get_parallel_exec_option()
self.taskIdsToCancel = set()
self.cancelEvent = threading.Event()
+ self.recovery_command_lock = threading.RLock()
+ self.active_recovery_task_ids = set()
self.component_status_executor =
initializer_module.component_status_executor
if self.parallel_execution == 1:
logger.info(
@@ -88,22 +90,69 @@ class ActionQueue(threading.Thread):
self.lock = threading.Lock()
def put(self, commands):
- for command in commands:
- if "serviceName" not in command:
- command["serviceName"] = "null"
- if "clusterId" not in command:
- command["clusterId"] = "null"
+ commands = list(commands)
+ with self.recovery_command_lock:
+ if any(
+ command.get("commandType") == AgentCommand.execution for command in
commands
+ ):
+ self.preempt_recovery_commands()
- logger.info(
- "Adding {commandType} for role {role} for service {serviceName} of
cluster_id {clusterId} to the queue".format(
- **command
+ for command in commands:
+ if "serviceName" not in command:
+ command["serviceName"] = "null"
+ if "clusterId" not in command:
+ command["clusterId"] = "null"
+
+ logger.info(
+ "Adding {commandType} for role {role} for service {serviceName} of
cluster_id {clusterId} to the queue".format(
+ **command
+ )
)
+
+ if command["commandType"] == AgentCommand.background_execution:
+ self.backgroundCommandQueue.put(self.create_command_handle(command))
+ else:
+ self.commandQueue.put(command)
+
+ def preempt_recovery_commands(self):
+ """Stop hidden recovery work so server-issued commands can run
immediately."""
+ queued_recovery_task_ids = []
+ with self.commandQueue.mutex:
+ retained_commands = []
+ for command in self.commandQueue.queue:
+ if (
+ command is not None
+ and command.get("commandType") == AgentCommand.auto_execution
+ ):
+ queued_recovery_task_ids.append(command["taskId"])
+ else:
+ retained_commands.append(command)
+ self.commandQueue.queue.clear()
+ self.commandQueue.queue.extend(retained_commands)
+
+ active_recovery_task_ids = list(self.active_recovery_task_ids)
+ if queued_recovery_task_ids or active_recovery_task_ids:
+ logger.info(
+ "Preempting auto recovery for server-issued commands. Queued task IDs:
%s; active task IDs: %s",
+ queued_recovery_task_ids,
+ active_recovery_task_ids,
)
- if command["commandType"] == AgentCommand.background_execution:
- self.backgroundCommandQueue.put(self.create_command_handle(command))
- else:
- self.commandQueue.put(command)
+ reason = "Preempted by a server-issued command"
+ for task_id in active_recovery_task_ids:
+ self.taskIdsToCancel.add(task_id)
+ self.customServiceOrchestrator.cancel_command(task_id, reason)
+
+ if active_recovery_task_ids:
+ self.cancelEvent.set()
+
+ def has_queued_server_command(self):
+ with self.commandQueue.mutex:
+ return any(
+ command is not None
+ and command.get("commandType") == AgentCommand.execution
+ for command in self.commandQueue.queue
+ )
def interrupt(self):
self.commandQueue.put(None)
@@ -186,8 +235,9 @@ class ActionQueue(threading.Thread):
logger.info("ActionQueue thread has successfully finished")
def fill_recovery_commands(self):
- if self.recovery_manager.enabled() and not
self.tasks_in_progress_or_pending():
- self.put(self.recovery_manager.get_recovery_commands())
+ with self.recovery_command_lock:
+ if self.recovery_manager.enabled() and not
self.tasks_in_progress_or_pending():
+ self.put(self.recovery_manager.get_recovery_commands())
def process_background_queue_safe_empty(self):
while not self.backgroundCommandQueue.empty():
@@ -211,6 +261,17 @@ class ActionQueue(threading.Thread):
# make sure we log failures
command_type = command["commandType"]
logger.debug("Took an element of Queue (command type = %s).", command_type)
+ is_recovery_command = command_type == AgentCommand.auto_execution
+ if is_recovery_command:
+ with self.recovery_command_lock:
+ if self.has_queued_server_command():
+ logger.info(
+ "Skipping auto recovery task %s because a server-issued command is
queued",
+ command["taskId"],
+ )
+ return
+ self.active_recovery_task_ids.add(command["taskId"])
+
try:
if command_type in AgentCommand.AUTO_EXECUTION_COMMAND_GROUP:
try:
@@ -226,6 +287,11 @@ class ActionQueue(threading.Thread):
logger.error("Unrecognized command %s", pprint.pformat(command))
except Exception:
logger.exception(f"Exception while processing {command_type} command")
+ finally:
+ if is_recovery_command:
+ with self.recovery_command_lock:
+ self.active_recovery_task_ids.discard(command["taskId"])
+ self.taskIdsToCancel.discard(command["taskId"])
def tasks_in_progress_or_pending(self):
return not self.commandQueue.empty() or
self.recovery_manager.has_active_command()
@@ -244,7 +310,8 @@ class ActionQueue(threading.Thread):
delay = 1
log_command_output = True
command_canceled = False
- command_result = {}
+ status = CommandStatus.failed
+ command_result = {"stdout": "", "stderr": "", "exitcode": -signal.SIGTERM}
message = (
"Executing command with id = {commandId}, taskId = {taskId} for role =
{role} of "
@@ -315,7 +382,8 @@ class ActionQueue(threading.Thread):
self.cancelEvent.clear()
# for case of command reschedule (e.g. command and cancel for the same
taskId are send at the same time)
- self.taskIdsToCancel.discard(taskId)
+ if command_type != AgentCommand.auto_execution:
+ self.taskIdsToCancel.discard(taskId)
while retry_duration >= 0:
if taskId in self.taskIdsToCancel:
diff --git
a/ambari-agent/src/main/python/ambari_agent/ComponentStatusExecutor.py
b/ambari-agent/src/main/python/ambari_agent/ComponentStatusExecutor.py
index d0b3955d11..8c868b83d4 100644
--- a/ambari-agent/src/main/python/ambari_agent/ComponentStatusExecutor.py
+++ b/ambari-agent/src/main/python/ambari_agent/ComponentStatusExecutor.py
@@ -49,6 +49,8 @@ class ComponentStatusExecutor(threading.Thread):
self.reports_to_discard = []
self.reports_to_discard_lock = threading.RLock()
self.reported_component_status_lock = threading.RLock()
+ self.component_status_snapshot_complete = False
+ self.component_status_snapshot_generation = 0
threading.Thread.__init__(self)
def run(self):
@@ -65,40 +67,52 @@ class ComponentStatusExecutor(threading.Thread):
try:
self.clean_not_existing_clusters_info()
cluster_reports = defaultdict(lambda: [])
+ cluster_ids = self.topology_cache.get_cluster_ids()
+ with self.reported_component_status_lock:
+ snapshot_generation = self.component_status_snapshot_generation
+ snapshot_complete = (
+ not self.component_status_snapshot_complete and bool(cluster_ids)
+ )
with self.reports_to_discard_lock:
self.reports_to_discard = []
- for cluster_id in self.topology_cache.get_cluster_ids():
+ for cluster_id in cluster_ids:
# TODO: check if we can make clusters immutable too
try:
topology_cache = self.topology_cache[cluster_id]
metadata_cache = self.metadata_cache[cluster_id]
except KeyError:
# multithreading: if cluster was deleted during iteration
+ snapshot_complete = False
continue
if "status_commands_to_run" not in metadata_cache:
+ snapshot_complete = False
continue
status_commands_to_run = metadata_cache.status_commands_to_run
if "components" not in topology_cache:
+ snapshot_complete = False
continue
current_host_id = self.topology_cache.get_current_host_id(cluster_id)
if current_host_id is None:
+ snapshot_complete = False
continue
cluster_components = topology_cache.components
for component_dict in cluster_components:
for command_name in status_commands_to_run:
if self.stop_event.is_set():
+ snapshot_complete = False
break
# cluster was already removed
if cluster_id not in self.topology_cache.get_cluster_ids():
+ snapshot_complete = False
break
# check if component is installed on current host
@@ -115,6 +129,7 @@ class ComponentStatusExecutor(threading.Thread):
self.logger.info(
f"Skipping status command for {component_name}. Since
command for it is running"
)
+ snapshot_complete = False
continue
result = self.check_component_status(
@@ -125,7 +140,11 @@ class ComponentStatusExecutor(threading.Thread):
cluster_reports[cluster_id].append(result)
cluster_reports = self.discard_stale_reports(cluster_reports)
- self.send_updates_to_server(cluster_reports)
+ self.send_updates_to_server(
+ cluster_reports,
+ snapshot_complete=snapshot_complete,
+ snapshot_generation=snapshot_generation,
+ )
except (
ConnectionIsAlreadyClosed
): # server and agent disconnected during sending data. Not an issue
@@ -212,12 +231,12 @@ class ComponentStatusExecutor(threading.Thread):
"clusterId": cluster_id,
}
- if (
- status
- !=
self.reported_component_status[cluster_id][f"{service_name}/{component_name}"][
- command_name
- ]
- ):
+ with self.reported_component_status_lock:
+ previous_status = self.reported_component_status[cluster_id][
+ f"{service_name}/{component_name}"
+ ][command_name]
+
+ if status != previous_status:
logging.info(f"Status for {component_name} has changed to {status}")
self.recovery_manager.handle_status_change(component_name, status)
@@ -257,22 +276,61 @@ class ComponentStatusExecutor(threading.Thread):
cluster_reports[cluster_id].append(report)
- self.send_updates_to_server(cluster_reports)
+ self.component_status_snapshot_complete = False
+ self.component_status_snapshot_generation += 1
+ self.reported_component_status.clear()
+
+ self.send_updates_to_server(
+ cluster_reports, snapshot_complete=False, save_reported_status=False
+ )
- def send_updates_to_server(self, cluster_reports):
- if not cluster_reports or not self.initializer_module.is_registered:
+ def send_updates_to_server(
+ self,
+ cluster_reports,
+ snapshot_complete=False,
+ save_reported_status=True,
+ snapshot_generation=None,
+ ):
+ if (
+ (not cluster_reports and not snapshot_complete and save_reported_status)
+ or not self.initializer_module.is_registered
+ ):
return
- correlation_id = self.initializer_module.connection.send(
- message={"clusters": cluster_reports},
- destination=Constants.COMPONENT_STATUS_REPORTS_ENDPOINT,
- )
-
self.server_responses_listener.listener_functions_on_success[correlation_id] = (
- lambda headers, message:
self.save_reported_component_status(cluster_reports)
- )
+ with self.reported_component_status_lock:
+ if snapshot_generation is None:
+ snapshot_generation = self.component_status_snapshot_generation
+ elif snapshot_generation != self.component_status_snapshot_generation:
+ return
+
+ correlation_id = self.initializer_module.connection.send(
+ message={
+ "clusters": cluster_reports,
+ "snapshotComplete": snapshot_complete,
+ },
+ destination=Constants.COMPONENT_STATUS_REPORTS_ENDPOINT,
+ )
+ if save_reported_status:
+
self.server_responses_listener.listener_functions_on_success[correlation_id] = (
+ lambda headers, message: self.save_reported_component_status(
+ cluster_reports, snapshot_complete, snapshot_generation
+ )
+ )
+ else:
+
self.server_responses_listener.listener_functions_on_success[correlation_id] = (
+ lambda headers, message: None
+ )
- def save_reported_component_status(self, cluster_reports):
+ def save_reported_component_status(
+ self, cluster_reports, snapshot_complete=False, snapshot_generation=None
+ ):
with self.reported_component_status_lock:
+ if (
+ snapshot_generation is not None
+ and snapshot_generation != self.component_status_snapshot_generation
+ ):
+ return
+
for cluster_id, reports in cluster_reports.items():
for report in reports:
component_name = report["componentName"]
@@ -283,6 +341,8 @@ class ComponentStatusExecutor(threading.Thread):
self.reported_component_status[cluster_id][
f"{service_name}/{component_name}"
][command] = status
+ if snapshot_complete:
+ self.component_status_snapshot_complete = True
def clean_not_existing_clusters_info(self):
"""
diff --git a/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
b/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
index a8bb70740c..5525c577ba 100644
--- a/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
+++ b/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
@@ -208,12 +208,13 @@ class HeartbeatThread(threading.Thread):
self.run_post_registration_actions()
- self.initializer_module.is_registered = True
+ # Invalidate cached status before the periodic scanner can use the new
session.
+ self.force_component_status_update()
# now when registration is done we can expose connection to other threads.
self.initializer_module._connection = self.connection
+ self.initializer_module.is_registered = True
self.report_components_initial_versions()
- self.force_component_status_update()
def run_post_registration_actions(self):
for post_registration_action in self.post_registration_actions:
diff --git a/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
b/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
index 0a2e693da0..43663190c8 100644
--- a/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
+++ b/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
@@ -98,6 +98,11 @@ class RecoveryManager:
self.allowed_desired_states = [self.STARTED, self.INSTALLED]
self.allowed_current_states = [self.INIT, self.INSTALLED]
self.enabled_components = []
+ self.component_dependencies = {}
+ self.topology_managed = False
+ self.topology_epoch = None
+ self.topology_version = -1
+ self.topology_complete = True
self.statuses = {}
self.__component_to_service_map = {} # component => service map TODO: fix
it later(hack here)
self.__status_lock = threading.RLock()
@@ -374,7 +379,7 @@ class RecoveryManager:
elif status["current"] == self.STARTED:
command = self.get_restart_command(component)
- if command:
+ if command and self.recovery_topology_allows(command):
self.execute(component)
logger.info(
"Created recovery command %s for component %s",
@@ -385,6 +390,54 @@ class RecoveryManager:
return commands
+ def recovery_topology_allows(self, command):
+ if not self.topology_managed:
+ return True
+
+ if not self.topology_complete:
+ logger.info(
+ "Recovery for %s is blocked until the component status topology is
complete",
+ command[self.ROLE],
+ )
+ return False
+
+ is_start = command[self.ROLE_COMMAND] == RoleCommand.start
+ is_restart = (
+ command[self.ROLE_COMMAND] == RoleCommand.custom_command
+ and command.get("custom_command") == CustomCommand.restart
+ )
+ if not is_start and not is_restart:
+ return True
+
+ for dependency in self.component_dependencies.get(command[self.ROLE], []):
+ required_state = dependency.get("required_state", self.STARTED)
+ if (
+ not dependency.get("fresh", False)
+ or not self.dependency_state_satisfies(
+ dependency.get("current_state"), required_state
+ )
+ or not self.dependency_state_satisfies(
+ dependency.get("desired_state"), required_state
+ )
+ ):
+ logger.info(
+ "Recovery for %s is blocked by %s on %s: current=%s, desired=%s,
required=%s, fresh=%s",
+ command[self.ROLE],
+ dependency.get("component_name"),
+ dependency.get("host_name"),
+ dependency.get("current_state"),
+ dependency.get("desired_state"),
+ required_state,
+ dependency.get("fresh", False),
+ )
+ return False
+ return True
+
+ def dependency_state_satisfies(self, state, required_state):
+ if required_state == self.INSTALLED:
+ return state in (self.INSTALLED, self.STARTED)
+ return state == required_state
+
def may_execute(self, action):
"""
Check if an action can be executed
@@ -554,9 +607,34 @@ class RecoveryManager:
if logger.isEnabledFor(logging.INFO):
logger.info("RecoverConfig = %s",
pprint.pformat(dictionary["recoveryConfig"]))
config = dictionary["recoveryConfig"]
+ if "topology_epoch" in config and "topology_version" in config:
+ topology_epoch = config["topology_epoch"]
+ topology_version = int(config["topology_version"])
+ if (
+ topology_epoch == self.topology_epoch
+ and topology_version < self.topology_version
+ ):
+ logger.warning(
+ "Ignoring stale recovery topology version %s; current version is
%s",
+ topology_version,
+ self.topology_version,
+ )
+ return
+
+ self.topology_managed = True
+ self.topology_epoch = topology_epoch
+ self.topology_version = topology_version
+ self.topology_complete = config.get("topology_complete", False)
+ else:
+ self.topology_managed = False
+ self.topology_epoch = None
+ self.topology_version = -1
+ self.topology_complete = True
+
if "components" in config:
enabled_components = config["components"]
enabled_components_list = []
+ component_dependencies = {}
components = [
(item["service_name"], item["component_name"], item["desired_state"])
@@ -572,7 +650,13 @@ class RecoveryManager:
# push another service <-> component relation
self.__component_to_service_map[component] = service
+ for item in enabled_components:
+ component_dependencies[item["component_name"]] = item.get(
+ "dependencies", []
+ )
+
self.enabled_components = enabled_components_list
+ self.component_dependencies = component_dependencies
def on_config_update(self):
recovery_enabled = False
diff --git a/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py
b/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py
index 032c1aaa8d..32ec1dae7c 100644
--- a/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py
+++ b/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py
@@ -64,6 +64,14 @@ class TestActionQueue(TestCase):
logger = logging.getLogger()
+ def create_mock_action_queue(self):
+ initializer_module = MagicMock()
+ initializer_module.config.get.return_value = "/tmp"
+ initializer_module.config.get_parallel_exec_option.return_value = 0
+ initializer_module.stop_event = threading.Event()
+ initializer_module.recovery_manager.enabled.return_value = False
+ return ActionQueue(initializer_module)
+
datanode_install_command = {
"commandType": "EXECUTION_COMMAND",
"role": "DATANODE",
@@ -108,6 +116,83 @@ class TestActionQueue(TestCase):
"clusterId": CLUSTER_ID,
}
+ def test_server_command_removes_queued_recovery_commands(self):
+ action_queue = self.create_mock_action_queue()
+ recovery_command = copy.deepcopy(self.datanode_auto_start_command)
+ server_command = copy.deepcopy(self.namenode_install_command)
+
+ action_queue.put([recovery_command])
+ action_queue.put([server_command])
+
+ self.assertEqual(1, action_queue.commandQueue.qsize())
+ self.assertEqual(server_command, action_queue.commandQueue.get_nowait())
+
+ def test_server_command_cancels_active_recovery_commands(self):
+ action_queue = self.create_mock_action_queue()
+ recovery_command = copy.deepcopy(self.datanode_auto_start_command)
+ server_command = copy.deepcopy(self.namenode_install_command)
+ execution_started = threading.Event()
+ allow_execution_to_finish = threading.Event()
+
+ def wait_for_preemption(command):
+ execution_started.set()
+ allow_execution_to_finish.wait(5)
+
+ action_queue.execute_command = MagicMock(side_effect=wait_for_preemption)
+ recovery_thread = threading.Thread(
+ target=action_queue.process_command, args=(recovery_command,)
+ )
+ recovery_thread.start()
+ self.assertTrue(execution_started.wait(5))
+
+ action_queue.put([server_command])
+
+
action_queue.customServiceOrchestrator.cancel_command.assert_called_once_with(
+ recovery_command["taskId"], "Preempted by a server-issued command"
+ )
+ self.assertIn(recovery_command["taskId"], action_queue.taskIdsToCancel)
+ self.assertEqual(server_command, action_queue.commandQueue.get_nowait())
+
+ allow_execution_to_finish.set()
+ recovery_thread.join(5)
+ self.assertFalse(recovery_thread.is_alive())
+ self.assertNotIn(recovery_command["taskId"], action_queue.taskIdsToCancel)
+
+ def test_dequeued_recovery_command_yields_to_queued_server_command(self):
+ action_queue = self.create_mock_action_queue()
+ recovery_command = copy.deepcopy(self.datanode_auto_start_command)
+ server_command = copy.deepcopy(self.namenode_install_command)
+ action_queue.commandQueue.put(server_command)
+ action_queue.execute_command = MagicMock()
+
+ action_queue.process_command(recovery_command)
+
+ action_queue.execute_command.assert_not_called()
+ self.assertNotIn(recovery_command["taskId"],
action_queue.active_recovery_task_ids)
+
+ def test_recovery_command_canceled_before_script_execution(self):
+ action_queue = self.create_mock_action_queue()
+ recovery_command = copy.deepcopy(self.datanode_auto_start_command)
+ action_queue.taskIdsToCancel.add(recovery_command["taskId"])
+ action_queue.commandStatuses.generate_report_template.return_value = {}
+ action_queue.config.get.side_effect = (
+ lambda section, key: "0" if key == "log_command_executes" else "/tmp"
+ )
+
+ action_queue.execute_command(recovery_command)
+
+ action_queue.customServiceOrchestrator.runCommand.assert_not_called()
+ self.assertNotIn(recovery_command["taskId"], action_queue.taskIdsToCancel)
+
+ def test_recovery_command_does_not_preempt_active_recovery(self):
+ action_queue = self.create_mock_action_queue()
+ active_task_id = 41
+ action_queue.active_recovery_task_ids.add(active_task_id)
+
+ action_queue.put([copy.deepcopy(self.datanode_auto_start_command)])
+
+ action_queue.customServiceOrchestrator.cancel_command.assert_not_called()
+
datanode_upgrade_command = {
"commandId": 17,
"role": "role",
diff --git
a/ambari-agent/src/test/python/ambari_agent/TestComponentStatusExecutor.py
b/ambari-agent/src/test/python/ambari_agent/TestComponentStatusExecutor.py
new file mode 100644
index 0000000000..092769d602
--- /dev/null
+++ b/ambari-agent/src/test/python/ambari_agent/TestComponentStatusExecutor.py
@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import threading
+from unittest import TestCase
+
+from ambari_agent.ComponentStatusExecutor import ComponentStatusExecutor
+from ambari_agent import Constants
+from mock.mock import MagicMock
+
+
+class TestComponentStatusExecutor(TestCase):
+ def create_executor(self):
+ initializer_module = MagicMock()
+ initializer_module.config.status_commands_run_interval = 10
+ initializer_module.stop_event = threading.Event()
+ initializer_module.is_registered = True
+ initializer_module.server_responses_listener.listener_functions_on_success
= {}
+ initializer_module.connection.send.return_value = "correlation-1"
+ return ComponentStatusExecutor(initializer_module), initializer_module
+
+ def test_complete_snapshot_is_marked_only_after_server_ack(self):
+ executor, initializer_module = self.create_executor()
+ reports = {
+ "1": [
+ {
+ "serviceName": "HDFS",
+ "componentName": "NAMENODE",
+ "command": "STATUS",
+ "status": "STARTED",
+ "clusterId": "1",
+ }
+ ]
+ }
+
+ executor.send_updates_to_server(reports, snapshot_complete=True)
+
+ initializer_module.connection.send.assert_called_once_with(
+ message={"clusters": reports, "snapshotComplete": True},
+ destination=Constants.COMPONENT_STATUS_REPORTS_ENDPOINT,
+ )
+ self.assertFalse(executor.component_status_snapshot_complete)
+
+ callback =
initializer_module.server_responses_listener.listener_functions_on_success[
+ "correlation-1"
+ ]
+ callback({}, {})
+ self.assertTrue(executor.component_status_snapshot_complete)
+
+ def test_ack_from_snapshot_before_forced_refresh_is_ignored(self):
+ executor, initializer_module = self.create_executor()
+ initializer_module.connection.send.side_effect = ["scan-1",
"forced-refresh"]
+ reports = {
+ "1": [
+ {
+ "serviceName": "HDFS",
+ "componentName": "NAMENODE",
+ "command": "STATUS",
+ "status": "STARTED",
+ "clusterId": "1",
+ }
+ ]
+ }
+
+ executor.send_updates_to_server(reports, snapshot_complete=True)
+ stale_callback = (
+
initializer_module.server_responses_listener.listener_functions_on_success[
+ "scan-1"
+ ]
+ )
+ executor.force_send_component_statuses()
+ stale_callback({}, {})
+
+ self.assertFalse(executor.component_status_snapshot_complete)
+ self.assertEqual({}, dict(executor.reported_component_status))
diff --git a/ambari-agent/src/test/python/ambari_agent/TestRecoveryManager.py
b/ambari-agent/src/test/python/ambari_agent/TestRecoveryManager.py
index 1a098a028c..ff9f4759de 100644
--- a/ambari-agent/src/test/python/ambari_agent/TestRecoveryManager.py
+++ b/ambari-agent/src/test/python/ambari_agent/TestRecoveryManager.py
@@ -563,6 +563,147 @@ class _TestRecoveryManager(TestCase):
self.assertFalse(rm.configured_for_recovery("E"))
self.assertTrue(rm.configured_for_recovery("F"))
+ def test_topology_requires_fresh_started_dependencies(self):
+ rm = RecoveryManager(MagicMock())
+ rm.update_config(12, 5, 1, 15, True, True, False)
+ component = {
+ "component_name": "RESOURCEMANAGER",
+ "service_name": "YARN",
+ "desired_state": "STARTED",
+ "dependencies": [
+ {
+ "component_name": "NAMENODE",
+ "service_name": "HDFS",
+ "host_name": "host1",
+ "current_state": "INSTALLED",
+ "desired_state": "STARTED",
+ "required_state": "STARTED",
+ "fresh": True,
+ }
+ ],
+ }
+
+ rm.update_recovery_config(
+ {
+ "recoveryConfig": {
+ "topology_epoch": "server-1",
+ "topology_version": 1,
+ "topology_complete": False,
+ "components": [component],
+ }
+ }
+ )
+ rm.update_current_status("RESOURCEMANAGER", "INSTALLED")
+ self.assertEqual([], rm.get_recovery_commands())
+
+ rm.update_recovery_config(
+ {
+ "recoveryConfig": {
+ "topology_epoch": "server-1",
+ "topology_version": 2,
+ "topology_complete": True,
+ "components": [component],
+ }
+ }
+ )
+ self.assertEqual([], rm.get_recovery_commands())
+
+ component["dependencies"][0]["current_state"] = "STARTED"
+ rm.update_recovery_config(
+ {
+ "recoveryConfig": {
+ "topology_epoch": "server-1",
+ "topology_version": 3,
+ "topology_complete": True,
+ "components": [component],
+ }
+ }
+ )
+ commands = rm.get_recovery_commands()
+ self.assertEqual(1, len(commands))
+ self.assertEqual("RESOURCEMANAGER", commands[0]["role"])
+ self.assertEqual("START", commands[0]["roleCommand"])
+
+ def test_recovery_config_rejects_older_topology_version(self):
+ rm = RecoveryManager(MagicMock())
+ rm.update_recovery_config(
+ {
+ "recoveryConfig": {
+ "topology_epoch": "server-1",
+ "topology_version": 5,
+ "topology_complete": True,
+ "components": [
+ {
+ "component_name": "NODEMANAGER",
+ "service_name": "YARN",
+ "desired_state": "STARTED",
+ }
+ ],
+ }
+ }
+ )
+ rm.update_recovery_config(
+ {
+ "recoveryConfig": {
+ "topology_epoch": "server-1",
+ "topology_version": 4,
+ "topology_complete": False,
+ "components": [
+ {
+ "component_name": "NODEMANAGER",
+ "service_name": "YARN",
+ "desired_state": "INSTALLED",
+ }
+ ],
+ }
+ }
+ )
+
+ self.assertEqual(5, rm.topology_version)
+ self.assertTrue(rm.topology_complete)
+ self.assertEqual("STARTED", rm.get_desired_status("NODEMANAGER"))
+
+ rm.update_recovery_config(
+ {
+ "recoveryConfig": {
+ "topology_epoch": "server-2",
+ "topology_version": 1,
+ "topology_complete": False,
+ "components": [],
+ }
+ }
+ )
+ self.assertEqual("server-2", rm.topology_epoch)
+ self.assertEqual(1, rm.topology_version)
+ self.assertFalse(rm.topology_complete)
+
+ def test_started_dependency_satisfies_installed_requirement(self):
+ rm = RecoveryManager(MagicMock())
+ rm.topology_managed = True
+ rm.topology_complete = True
+ rm.component_dependencies = {
+ "RESOURCEMANAGER": [
+ {
+ "component_name": "NAMENODE",
+ "service_name": "HDFS",
+ "host_name": "host1",
+ "current_state": "STARTED",
+ "desired_state": "STARTED",
+ "required_state": "INSTALLED",
+ "fresh": True,
+ }
+ ]
+ }
+
+ self.assertTrue(
+ rm.recovery_topology_allows(
+ {
+ "role": "RESOURCEMANAGER",
+ "roleCommand": "START",
+ }
+ )
+ )
+
@patch.object(RecoveryManager, "_now_")
def test_reset_if_window_passed_since_last_attempt(self, time_mock):
time_mock.side_effect = [1000, 1071, 1372]
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/ComponentStatusAgentReport.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/ComponentStatusAgentReport.java
index e6a3813190..0bc0aa4da0 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/ComponentStatusAgentReport.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/ComponentStatusAgentReport.java
@@ -20,17 +20,41 @@ package org.apache.ambari.server.agent;
import java.util.List;
import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.agent.stomp.HostLevelParamsHolder;
public class ComponentStatusAgentReport extends
AgentReport<List<ComponentStatus>> {
private final HeartBeatHandler hh;
+ private final HostLevelParamsHolder hostLevelParamsHolder;
+ private final RecoveryTopologyManager recoveryTopologyManager;
+ private final Long hostId;
+ private final String sessionId;
+ private final boolean snapshotComplete;
- public ComponentStatusAgentReport(HeartBeatHandler hh, String hostName,
List<ComponentStatus> componentStatuses) {
+ public ComponentStatusAgentReport(HeartBeatHandler hh, String hostName,
List<ComponentStatus> componentStatuses,
+ HostLevelParamsHolder hostLevelParamsHolder, RecoveryTopologyManager
recoveryTopologyManager,
+ Long hostId, String sessionId, boolean snapshotComplete) {
super(hostName, componentStatuses);
this.hh = hh;
+ this.hostLevelParamsHolder = hostLevelParamsHolder;
+ this.recoveryTopologyManager = recoveryTopologyManager;
+ this.hostId = hostId;
+ this.sessionId = sessionId;
+ this.snapshotComplete = snapshotComplete;
}
@Override
protected void process(List<ComponentStatus> report, String hostName) throws
AmbariException {
+ if (!recoveryTopologyManager.isActiveSession(hostId, sessionId)) {
+ return;
+ }
+
hh.handleComponentReportStatus(report, hostName);
+ if (!report.isEmpty() || snapshotComplete) {
+ recoveryTopologyManager.componentStateUpdated();
+ if (snapshotComplete) {
+ recoveryTopologyManager.markSnapshotComplete(hostId, sessionId);
+ }
+ hostLevelParamsHolder.updateRecoveryTopology(hostName);
+ }
}
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
index 007e49ceef..d7cf416b95 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/HeartbeatMonitor.java
@@ -37,6 +37,7 @@ import java.util.TreeMap;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.actionmanager.ActionManager;
+import org.apache.ambari.server.agent.stomp.HostLevelParamsHolder;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AmbariManagementController;
@@ -81,6 +82,8 @@ public class HeartbeatMonitor implements Runnable {
private final Configuration configuration;
private final AgentRequests agentRequests;
private final AmbariEventPublisher ambariEventPublisher;
+ private final HostLevelParamsHolder hostLevelParamsHolder;
+ private final RecoveryTopologyManager recoveryTopologyManager;
public HeartbeatMonitor(Clusters clusters, ActionManager am,
int threadWakeupInterval, Injector injector) {
@@ -94,6 +97,8 @@ public class HeartbeatMonitor implements Runnable {
configuration = injector.getInstance(Configuration.class);
agentRequests = new AgentRequests();
ambariEventPublisher = injector.getInstance(AmbariEventPublisher.class);
+ hostLevelParamsHolder = injector.getInstance(HostLevelParamsHolder.class);
+ recoveryTopologyManager =
injector.getInstance(RecoveryTopologyManager.class);
ambariEventPublisher.register(this);
}
@@ -346,6 +351,9 @@ public class HeartbeatMonitor implements Runnable {
}
}
+ recoveryTopologyManager.endAgentSession(hostId);
+ hostLevelParamsHolder.updateRecoveryTopology(host);
+
//Purge action queue
//notify action manager
actionManager.handleLostHost(host);
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfig.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfig.java
index e71d9ccc88..a71a17b096 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfig.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfig.java
@@ -33,14 +33,46 @@ public class RecoveryConfig {
@JsonProperty("components")
private List<RecoveryConfigComponent> enabledComponents;
+ @SerializedName("topology_epoch")
+ @JsonProperty("topology_epoch")
+ private String topologyEpoch;
+
+ @SerializedName("topology_version")
+ @JsonProperty("topology_version")
+ private long topologyVersion;
+
+ @SerializedName("topology_complete")
+ @JsonProperty("topology_complete")
+ private boolean topologyComplete;
+
public RecoveryConfig(List<RecoveryConfigComponent> enabledComponents) {
+ this(enabledComponents, null, 0, true);
+ }
+
+ public RecoveryConfig(List<RecoveryConfigComponent> enabledComponents,
String topologyEpoch,
+ long topologyVersion, boolean topologyComplete) {
this.enabledComponents = enabledComponents;
+ this.topologyEpoch = topologyEpoch;
+ this.topologyVersion = topologyVersion;
+ this.topologyComplete = topologyComplete;
}
public List<RecoveryConfigComponent> getEnabledComponents() {
return enabledComponents == null ? null :
Collections.unmodifiableList(enabledComponents);
}
+ public String getTopologyEpoch() {
+ return topologyEpoch;
+ }
+
+ public long getTopologyVersion() {
+ return topologyVersion;
+ }
+
+ public boolean isTopologyComplete() {
+ return topologyComplete;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
@@ -48,12 +80,21 @@ public class RecoveryConfig {
RecoveryConfig that = (RecoveryConfig) o;
- return enabledComponents != null ?
enabledComponents.equals(that.enabledComponents) : that.enabledComponents ==
null;
+ if (topologyVersion != that.topologyVersion || topologyComplete !=
that.topologyComplete) {
+ return false;
+ }
+ if (enabledComponents != null ?
!enabledComponents.equals(that.enabledComponents) : that.enabledComponents !=
null) {
+ return false;
+ }
+ return topologyEpoch != null ? topologyEpoch.equals(that.topologyEpoch) :
that.topologyEpoch == null;
}
@Override
public int hashCode() {
int result = (enabledComponents != null ? enabledComponents.hashCode() :
0);
+ result = 31 * result + (topologyEpoch != null ? topologyEpoch.hashCode() :
0);
+ result = 31 * result + Long.hashCode(topologyVersion);
+ result = 31 * result + (topologyComplete ? 1 : 0);
return result;
}
@@ -61,6 +102,9 @@ public class RecoveryConfig {
public String toString() {
StringBuilder buffer = new StringBuilder("RecoveryConfig{");
buffer.append(", components=").append(enabledComponents);
+ buffer.append(", topologyEpoch=").append(topologyEpoch);
+ buffer.append(", topologyVersion=").append(topologyVersion);
+ buffer.append(", topologyComplete=").append(topologyComplete);
buffer.append('}');
return buffer.toString();
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigComponent.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigComponent.java
index 50f13b4ea2..29b2157c14 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigComponent.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigComponent.java
@@ -18,6 +18,8 @@
package org.apache.ambari.server.agent;
+import java.util.Collections;
+import java.util.List;
import java.util.Objects;
import org.apache.ambari.server.state.ServiceComponentHost;
@@ -43,15 +45,23 @@ public class RecoveryConfigComponent{
@JsonProperty("desired_state")
private String desiredState;
+ private List<RecoveryConfigDependency> dependencies;
+
/**
* Creates new instance of {@link RecoveryConfigComponent}
* @param componentName name of the component
* @param desiredState desired desiredState of the component
*/
public RecoveryConfigComponent(String componentName, String serviceName,
State desiredState){
+ this(componentName, serviceName, desiredState, Collections.emptyList());
+ }
+
+ public RecoveryConfigComponent(String componentName, String serviceName,
State desiredState,
+ List<RecoveryConfigDependency> dependencies) {
this.setComponentName(componentName);
this.setServiceName(serviceName);
this.setDesiredState(desiredState);
+ this.dependencies = dependencies;
}
/**
@@ -96,6 +106,14 @@ public class RecoveryConfigComponent{
this.serviceName = serviceName;
}
+ public List<RecoveryConfigDependency> getDependencies() {
+ return dependencies == null ? Collections.emptyList() :
Collections.unmodifiableList(dependencies);
+ }
+
+ public void setDependencies(List<RecoveryConfigDependency> dependencies) {
+ this.dependencies = dependencies;
+ }
+
@Override
public boolean equals(Object o){
if (this == o) {
@@ -108,11 +126,12 @@ public class RecoveryConfigComponent{
final RecoveryConfigComponent that = (RecoveryConfigComponent) o;
return Objects.equals(componentName, that.componentName) &&
Objects.equals(serviceName, that.serviceName) &&
- Objects.equals(desiredState, that.desiredState);
+ Objects.equals(desiredState, that.desiredState) &&
+ Objects.equals(dependencies, that.dependencies);
}
@Override
public int hashCode(){
- return Objects.hash(componentName, serviceName, desiredState);
+ return Objects.hash(componentName, serviceName, desiredState,
dependencies);
}
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigDependency.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigDependency.java
new file mode 100644
index 0000000000..abb5aa9de3
--- /dev/null
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigDependency.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent;
+
+import java.util.Objects;
+
+import org.apache.ambari.server.state.ServiceComponentHost;
+import org.apache.ambari.server.state.State;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.gson.annotations.SerializedName;
+
+/** Runtime state for one component instance that blocks recovery. */
+public class RecoveryConfigDependency {
+ @SerializedName("component_name")
+ @JsonProperty("component_name")
+ private final String componentName;
+
+ @SerializedName("service_name")
+ @JsonProperty("service_name")
+ private final String serviceName;
+
+ @SerializedName("host_name")
+ @JsonProperty("host_name")
+ private final String hostName;
+
+ @SerializedName("current_state")
+ @JsonProperty("current_state")
+ private final State currentState;
+
+ @SerializedName("desired_state")
+ @JsonProperty("desired_state")
+ private final State desiredState;
+
+ @SerializedName("required_state")
+ @JsonProperty("required_state")
+ private final State requiredState;
+
+ private final boolean fresh;
+
+ public RecoveryConfigDependency(ServiceComponentHost sch, State
requiredState, boolean fresh) {
+ componentName = sch.getServiceComponentName();
+ serviceName = sch.getServiceName();
+ hostName = sch.getHostName();
+ currentState = sch.getState();
+ desiredState = sch.getDesiredState();
+ this.requiredState = requiredState;
+ this.fresh = fresh;
+ }
+
+ public String getComponentName() {
+ return componentName;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public String getHostName() {
+ return hostName;
+ }
+
+ public State getCurrentState() {
+ return currentState;
+ }
+
+ public State getDesiredState() {
+ return desiredState;
+ }
+
+ public State getRequiredState() {
+ return requiredState;
+ }
+
+ public boolean isFresh() {
+ return fresh;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RecoveryConfigDependency that = (RecoveryConfigDependency) o;
+ return fresh == that.fresh && Objects.equals(componentName,
that.componentName)
+ && Objects.equals(serviceName, that.serviceName) &&
Objects.equals(hostName, that.hostName)
+ && currentState == that.currentState && desiredState ==
that.desiredState
+ && requiredState == that.requiredState;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(componentName, serviceName, hostName, currentState,
desiredState, requiredState, fresh);
+ }
+}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigHelper.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigHelper.java
index 75a88afb95..c60cdd5f44 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigHelper.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryConfigHelper.java
@@ -19,18 +19,24 @@
package org.apache.ambari.server.agent;
import java.util.ArrayList;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.Role;
+import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.events.ClusterConfigChangedEvent;
import org.apache.ambari.server.events.MaintenanceModeEvent;
import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
import org.apache.ambari.server.events.ServiceComponentRecoveryChangedEvent;
import org.apache.ambari.server.events.ServiceComponentUninstalledEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
+import org.apache.ambari.server.metadata.RoleCommandOrder;
+import org.apache.ambari.server.metadata.RoleCommandOrderProvider;
+import org.apache.ambari.server.metadata.RoleCommandPair;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Config;
@@ -39,6 +45,8 @@ import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.MaintenanceState;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponentHost;
+import org.apache.ambari.server.state.StackId;
+import org.apache.ambari.server.state.State;
import org.apache.commons.lang.StringUtils;
import com.google.common.eventbus.AllowConcurrentEvents;
@@ -66,6 +74,12 @@ public class RecoveryConfigHelper {
@Inject
private Clusters clusters;
+ @Inject
+ private RoleCommandOrderProvider roleCommandOrderProvider;
+
+ @Inject
+ private RecoveryTopologyManager recoveryTopologyManager;
+
/**
* Cluster --> Host --> Timestamp
*/
@@ -101,10 +115,75 @@ public class RecoveryConfigHelper {
AutoStartConfig autoStartConfig = new AutoStartConfig(clusterName);
- RecoveryConfig recoveryConfig = new
RecoveryConfig(autoStartConfig.getEnabledComponents(hostname));
+ List<RecoveryConfigComponent> enabledComponents =
autoStartConfig.getEnabledComponents(hostname);
+ boolean topologyComplete = addTopologyState(autoStartConfig.cluster,
hostname, enabledComponents);
+ RecoveryConfig recoveryConfig = new RecoveryConfig(enabledComponents,
recoveryTopologyManager.getEpoch(),
+ recoveryTopologyManager.getVersion(), topologyComplete);
return recoveryConfig;
}
+ private boolean addTopologyState(Cluster cluster, String hostname,
+ List<RecoveryConfigComponent> enabledComponents) throws AmbariException {
+ if (cluster == null || StringUtils.isEmpty(hostname)) {
+ return false;
+ }
+
+ Host recoveryHost = clusters.getHost(hostname);
+ if (recoveryHost == null) {
+ return false;
+ }
+
+ boolean topologyComplete =
recoveryTopologyManager.isFresh(recoveryHost.getHostId());
+ if (enabledComponents.isEmpty()) {
+ return topologyComplete;
+ }
+
+ for (Service service : cluster.getServices().values()) {
+ StackId stackId = service.getDesiredStackId();
+ if (stackId == null) {
+ return false;
+ }
+ }
+
+ RoleCommandOrder roleCommandOrder =
roleCommandOrderProvider.getRoleCommandOrder(cluster);
+ if (roleCommandOrder == null) {
+ return false;
+ }
+
+ for (RecoveryConfigComponent component : enabledComponents) {
+ List<RecoveryConfigDependency> dependencies = new ArrayList<>();
+ try {
+ for (RoleCommandPair dependency : roleCommandOrder.getDependencies(
+ Role.valueOf(component.getComponentName()), RoleCommand.START)) {
+ State requiredState = getRequiredState(dependency.getCmd());
+ for (Service service : cluster.getServices().values()) {
+ if
(!service.getServiceComponents().containsKey(dependency.getRole().toString())) {
+ continue;
+ }
+ for (ServiceComponentHost dependencyHost :
service.getServiceComponent(
+
dependency.getRole().toString()).getServiceComponentHosts().values()) {
+ boolean fresh =
recoveryTopologyManager.isFresh(dependencyHost.getHost().getHostId());
+ dependencies.add(new RecoveryConfigDependency(dependencyHost,
requiredState, fresh));
+ topologyComplete &= fresh;
+ }
+ }
+ }
+ } catch (IllegalArgumentException e) {
+ topologyComplete = false;
+ }
+
+
dependencies.sort(Comparator.comparing(RecoveryConfigDependency::getServiceName)
+ .thenComparing(RecoveryConfigDependency::getComponentName)
+ .thenComparing(RecoveryConfigDependency::getHostName));
+ component.setDependencies(dependencies);
+ }
+ return topologyComplete;
+ }
+
+ private State getRequiredState(RoleCommand command) {
+ return command == RoleCommand.INSTALL ? State.INSTALLED : State.STARTED;
+ }
+
/**
* Computes if the recovery configuration was updated since the last time it
was sent to the agent.
*
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryTopologyManager.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryTopologyManager.java
new file mode 100644
index 0000000000..8a4aff05c8
--- /dev/null
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/RecoveryTopologyManager.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.google.inject.Singleton;
+
+/**
+ * Tracks the server lifecycle and the agent sessions that supplied fresh
component state.
+ */
+@Singleton
+public class RecoveryTopologyManager {
+ private final String epoch = UUID.randomUUID().toString();
+ private final AtomicLong version = new AtomicLong(1);
+ private final Map<Long, String> activeSessions = new ConcurrentHashMap<>();
+ private final Map<Long, String> completeSnapshotSessions = new
ConcurrentHashMap<>();
+
+ public void beginAgentSession(Long hostId, String sessionId) {
+ activeSessions.put(hostId, sessionId);
+ completeSnapshotSessions.remove(hostId);
+ version.incrementAndGet();
+ }
+
+ public boolean markSnapshotComplete(Long hostId, String sessionId) {
+ if (!sessionId.equals(activeSessions.get(hostId))) {
+ return false;
+ }
+
+ String previousSession = completeSnapshotSessions.put(hostId, sessionId);
+ if (!sessionId.equals(previousSession)) {
+ version.incrementAndGet();
+ return true;
+ }
+ return false;
+ }
+
+ public boolean isActiveSession(Long hostId, String sessionId) {
+ return sessionId.equals(activeSessions.get(hostId));
+ }
+
+ public void endAgentSession(Long hostId) {
+ activeSessions.remove(hostId);
+ completeSnapshotSessions.remove(hostId);
+ version.incrementAndGet();
+ }
+
+ public void componentStateUpdated() {
+ version.incrementAndGet();
+ }
+
+ public boolean isFresh(Long hostId) {
+ String activeSession = activeSessions.get(hostId);
+ return activeSession != null &&
activeSession.equals(completeSnapshotSessions.get(hostId));
+ }
+
+ public String getEpoch() {
+ return epoch;
+ }
+
+ public long getVersion() {
+ return version.get();
+ }
+}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java
index 2830b15590..521eccb376 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/AgentReportsController.java
@@ -36,6 +36,7 @@ import
org.apache.ambari.server.agent.ComponentStatusAgentReport;
import org.apache.ambari.server.agent.ComponentVersionAgentReport;
import org.apache.ambari.server.agent.HeartBeatHandler;
import org.apache.ambari.server.agent.HostStatusAgentReport;
+import org.apache.ambari.server.agent.RecoveryTopologyManager;
import org.apache.ambari.server.agent.stomp.dto.AckReport;
import org.apache.ambari.server.agent.stomp.dto.CommandStatusReports;
import org.apache.ambari.server.agent.stomp.dto.ComponentStatusReport;
@@ -44,6 +45,7 @@ import
org.apache.ambari.server.agent.stomp.dto.ComponentVersionReports;
import org.apache.ambari.server.agent.stomp.dto.HostStatusReport;
import org.apache.ambari.server.events.DefaultMessageEmitter;
import org.apache.ambari.server.state.Alert;
+import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,11 +69,15 @@ public class AgentReportsController {
private final HeartBeatHandler hh;
private final AgentSessionManager agentSessionManager;
private final AgentReportsProcessor agentReportsProcessor;
+ private final HostLevelParamsHolder hostLevelParamsHolder;
+ private final RecoveryTopologyManager recoveryTopologyManager;
public AgentReportsController(Injector injector) {
hh = injector.getInstance(HeartBeatHandler.class);
agentSessionManager = injector.getInstance(AgentSessionManager.class);
agentReportsProcessor = injector.getInstance(AgentReportsProcessor.class);
+ hostLevelParamsHolder = injector.getInstance(HostLevelParamsHolder.class);
+ recoveryTopologyManager =
injector.getInstance(RecoveryTopologyManager.class);
}
@MessageMapping("/component_version")
@@ -98,8 +104,10 @@ public class AgentReportsController {
}
}
- agentReportsProcessor.addAgentReport(new ComponentStatusAgentReport(hh,
- agentSessionManager.getHost(simpSessionId).getHostName(), statuses));
+ Host host = agentSessionManager.getHost(simpSessionId);
+ agentReportsProcessor.addAgentReport(new ComponentStatusAgentReport(hh,
host.getHostName(), statuses,
+ hostLevelParamsHolder, recoveryTopologyManager, host.getHostId(),
simpSessionId,
+ message.isSnapshotComplete()));
return new ReportsResponse();
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
index 0447160e82..78db2c1e7e 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/HeartbeatController.java
@@ -34,11 +34,13 @@ import org.apache.ambari.server.agent.AgentSessionManager;
import org.apache.ambari.server.agent.HeartBeat;
import org.apache.ambari.server.agent.HeartBeatHandler;
import org.apache.ambari.server.agent.HeartBeatResponse;
+import org.apache.ambari.server.agent.RecoveryTopologyManager;
import org.apache.ambari.server.agent.Register;
import org.apache.ambari.server.agent.RegistrationResponse;
import org.apache.ambari.server.agent.RegistrationStatus;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.configuration.spring.GuiceBeansConfig;
+import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.cluster.ClustersImpl;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
import org.slf4j.Logger;
@@ -63,6 +65,8 @@ public class HeartbeatController {
private final HeartBeatHandler hh;
private final ClustersImpl clusters;
private final AgentSessionManager agentSessionManager;
+ private final HostLevelParamsHolder hostLevelParamsHolder;
+ private final RecoveryTopologyManager recoveryTopologyManager;
private final LinkedBlockingQueue queue;
private final ThreadFactory threadFactoryExecutor = new
ThreadFactoryBuilder().setNameFormat("agent-register-processor-%d").build();
private final ThreadFactory threadFactoryTimeout = new
ThreadFactoryBuilder().setNameFormat("agent-register-timeout-%d").build();
@@ -78,6 +82,8 @@ public class HeartbeatController {
clusters = injector.getInstance(ClustersImpl.class);
unitOfWork = injector.getInstance(UnitOfWork.class);
agentSessionManager = injector.getInstance(AgentSessionManager.class);
+ hostLevelParamsHolder = injector.getInstance(HostLevelParamsHolder.class);
+ recoveryTopologyManager =
injector.getInstance(RecoveryTopologyManager.class);
Configuration configuration = injector.getInstance(Configuration.class);
queue = new
LinkedBlockingQueue(configuration.getAgentsRegistrationQueueSize());
@@ -98,8 +104,10 @@ public class HeartbeatController {
try {
/* Call into the heartbeat handler */
response = hh.handleRegistration(message);
- agentSessionManager.register(simpSessionId,
- clusters.getHost(message.getHostname()));
+ Host host = clusters.getHost(message.getHostname());
+ recoveryTopologyManager.beginAgentSession(host.getHostId(),
simpSessionId);
+ hostLevelParamsHolder.updateRecoveryTopology(message.getHostname());
+ agentSessionManager.register(simpSessionId, host);
LOG.debug("Sending registration response " + response);
} catch (Exception ex) {
LOG.info(ex.getMessage(), ex);
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 7674233830..3525ea6e1a 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
@@ -19,6 +19,7 @@ package org.apache.ambari.server.agent.stomp;
import java.util.HashMap;
import java.util.Map;
+import java.util.Objects;
import java.util.TreeMap;
import org.apache.ambari.server.AmbariException;
@@ -94,6 +95,14 @@ public class HostLevelParamsHolder extends
AgentHostDataHolder<HostLevelParamsUp
}
}
+ public void updateRecoveryTopology(String reportingHostName) throws
AmbariException {
+ for (Cluster cluster : clusters.getClustersForHost(reportingHostName)) {
+ for (Host host : cluster.getHosts()) {
+ updateDataOfHost(cluster.getClusterId(), cluster, host);
+ }
+ }
+ }
+
@Override
protected HostLevelParamsUpdateEvent handleUpdate(HostLevelParamsUpdateEvent
current, HostLevelParamsUpdateEvent update) {
HostLevelParamsUpdateEvent result = null;
@@ -117,7 +126,9 @@ public class HostLevelParamsHolder extends
AgentHostDataHolder<HostLevelParamsUp
RecoveryConfig mergedRecoveryConfig;
Map<String, BlueprintProvisioningState>
mergedBlueprintProvisioningStates;
- if
(!currentCluster.getRecoveryConfig().equals(updatedCluster.getRecoveryConfig()))
{
+ if (isOlderRecoveryTopology(currentCluster.getRecoveryConfig(),
updatedCluster.getRecoveryConfig())) {
+ mergedRecoveryConfig = currentCluster.getRecoveryConfig();
+ } else if
(!currentCluster.getRecoveryConfig().equals(updatedCluster.getRecoveryConfig()))
{
mergedRecoveryConfig = updatedCluster.getRecoveryConfig();
clusterChanged = true;
} else {
@@ -152,6 +163,11 @@ public class HostLevelParamsHolder extends
AgentHostDataHolder<HostLevelParamsUp
return result;
}
+ private boolean isOlderRecoveryTopology(RecoveryConfig current,
RecoveryConfig update) {
+ return Objects.equals(current.getTopologyEpoch(),
update.getTopologyEpoch())
+ && update.getTopologyVersion() < current.getTopologyVersion();
+ }
+
@Override
protected HostLevelParamsUpdateEvent getEmptyData() {
return HostLevelParamsUpdateEvent.emptyUpdate();
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java
index ac327d4ec3..783891a3c2 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/agent/stomp/dto/ComponentStatusReports.java
@@ -28,6 +28,8 @@ public class ComponentStatusReports {
@JsonProperty("clusters")
private TreeMap<String, List<ComponentStatusReport>> componentStatusReports;
+ private boolean snapshotComplete;
+
public ComponentStatusReports() {
}
@@ -42,4 +44,12 @@ public class ComponentStatusReports {
public void setComponentStatusReports(TreeMap<String,
List<ComponentStatusReport>> componentStatusReports) {
this.componentStatusReports = componentStatusReports;
}
+
+ public boolean isSnapshotComplete() {
+ return snapshotComplete;
+ }
+
+ public void setSnapshotComplete(boolean snapshotComplete) {
+ this.snapshotComplete = snapshotComplete;
+ }
}
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandOrder.java
b/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandOrder.java
index b748a2b4dc..da560df920 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandOrder.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandOrder.java
@@ -18,6 +18,7 @@
package org.apache.ambari.server.metadata;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
@@ -188,6 +189,14 @@ public class RoleCommandOrder implements Cloneable {
return 0;
}
+ /**
+ * Returns the transitive role dependencies used by the stage planner.
+ */
+ public Set<RoleCommandPair> getDependencies(Role role, RoleCommand command) {
+ Set<RoleCommandPair> roleDependencies = dependencies.get(new
RoleCommandPair(role, command));
+ return roleDependencies == null ? Collections.emptySet() :
Collections.unmodifiableSet(roleDependencies);
+ }
+
/**
* Returns transitive dependencies as a services list
* @param service to check if it depends on another services
diff --git
a/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandPair.java
b/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandPair.java
index dccddc65c6..1431b41174 100644
---
a/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandPair.java
+++
b/ambari-server/src/main/java/org/apache/ambari/server/metadata/RoleCommandPair.java
@@ -53,11 +53,11 @@ public class RoleCommandPair {
return false;
}
- Role getRole() {
+ public Role getRole() {
return role;
}
- RoleCommand getCmd() {
+ public RoleCommand getCmd() {
return cmd;
}
diff --git
a/ambari-server/src/test/java/org/apache/ambari/server/agent/ComponentStatusAgentReportTest.java
b/ambari-server/src/test/java/org/apache/ambari/server/agent/ComponentStatusAgentReportTest.java
new file mode 100644
index 0000000000..67237bcede
--- /dev/null
+++
b/ambari-server/src/test/java/org/apache/ambari/server/agent/ComponentStatusAgentReportTest.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+import java.util.Collections;
+
+import org.apache.ambari.server.agent.stomp.HostLevelParamsHolder;
+import org.junit.Test;
+
+public class ComponentStatusAgentReportTest {
+ @Test
+ public void testReportFromSupersededSessionIsIgnored() throws Exception {
+ HeartBeatHandler heartbeatHandler = mock(HeartBeatHandler.class);
+ HostLevelParamsHolder hostLevelParamsHolder =
mock(HostLevelParamsHolder.class);
+ RecoveryTopologyManager recoveryTopologyManager = new
RecoveryTopologyManager();
+ Long hostId = 1L;
+ recoveryTopologyManager.beginAgentSession(hostId, "current-session");
+
+ ComponentStatusAgentReport report = new
ComponentStatusAgentReport(heartbeatHandler, "host1",
+ Collections.emptyList(), hostLevelParamsHolder,
recoveryTopologyManager, hostId,
+ "superseded-session", true);
+
+ report.process();
+
+ verifyNoInteractions(heartbeatHandler, hostLevelParamsHolder);
+ }
+}
diff --git
a/ambari-server/src/test/java/org/apache/ambari/server/agent/RecoveryTopologyManagerTest.java
b/ambari-server/src/test/java/org/apache/ambari/server/agent/RecoveryTopologyManagerTest.java
new file mode 100644
index 0000000000..a2aeb96fea
--- /dev/null
+++
b/ambari-server/src/test/java/org/apache/ambari/server/agent/RecoveryTopologyManagerTest.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.agent;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+public class RecoveryTopologyManagerTest {
+ @Test
+ public void testSnapshotIsFreshOnlyForTheActiveAgentSession() {
+ RecoveryTopologyManager manager = new RecoveryTopologyManager();
+ Long hostId = 1L;
+
+ manager.beginAgentSession(hostId, "session-1");
+ assertFalse(manager.isFresh(hostId));
+ assertTrue(manager.markSnapshotComplete(hostId, "session-1"));
+ assertTrue(manager.isFresh(hostId));
+
+ manager.beginAgentSession(hostId, "session-2");
+ assertFalse(manager.isFresh(hostId));
+ assertFalse(manager.markSnapshotComplete(hostId, "session-1"));
+ assertFalse(manager.isFresh(hostId));
+ assertTrue(manager.markSnapshotComplete(hostId, "session-2"));
+ assertTrue(manager.isFresh(hostId));
+ }
+
+ @Test
+ public void testEndingAgentSessionInvalidatesSnapshotAndVersion() {
+ RecoveryTopologyManager manager = new RecoveryTopologyManager();
+ Long hostId = 1L;
+
+ manager.beginAgentSession(hostId, "session-1");
+ manager.markSnapshotComplete(hostId, "session-1");
+ long completeVersion = manager.getVersion();
+
+ manager.endAgentSession(hostId);
+
+ assertFalse(manager.isActiveSession(hostId, "session-1"));
+ assertFalse(manager.isFresh(hostId));
+ assertTrue(manager.getVersion() > completeVersion);
+ }
+}
diff --git
a/ambari-server/src/test/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolderTest.java
b/ambari-server/src/test/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolderTest.java
index 63c733f892..6d6fd87e46 100644
---
a/ambari-server/src/test/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolderTest.java
+++
b/ambari-server/src/test/java/org/apache/ambari/server/agent/stomp/HostLevelParamsHolderTest.java
@@ -108,4 +108,21 @@ public class HostLevelParamsHolderTest {
assertTrue(result.getHostLevelParamsClusters().containsKey("1"));
assertTrue(result.getHostLevelParamsClusters().containsKey("2"));
}
+
+ @Test
+ public void testHandleUpdateDoesNotRegressRecoveryTopology() {
+ RecoveryConfig currentRecoveryConfig = new RecoveryConfig(null,
"server-1", 2, true);
+ HostLevelParamsCluster currentCluster = new
HostLevelParamsCluster(currentRecoveryConfig, Collections.emptyMap());
+ HostLevelParamsUpdateEvent current = new
HostLevelParamsUpdateEvent(HOST_ID,
+ Collections.singletonMap("1", currentCluster));
+
+ RecoveryConfig olderRecoveryConfig = new RecoveryConfig(null, "server-1",
1, false);
+ HostLevelParamsCluster olderCluster = new
HostLevelParamsCluster(olderRecoveryConfig, Collections.emptyMap());
+ HostLevelParamsUpdateEvent update = new HostLevelParamsUpdateEvent(HOST_ID,
+ Collections.singletonMap("1", olderCluster));
+
+ HostLevelParamsHolder levelParamsHolder = new
HostLevelParamsHolder(createNiceMock(AmbariEventPublisher.class));
+
+ assertEquals(null, levelParamsHolder.handleUpdate(current, update));
+ }
}
diff --git
a/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
b/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
index db1155b391..a9b5d09186 100644
---
a/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
+++
b/ambari-server/src/test/java/org/apache/ambari/server/configuration/RecoveryConfigHelperTest.java
@@ -35,7 +35,9 @@ import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.agent.HeartbeatTestHelper;
import org.apache.ambari.server.agent.RecoveryConfig;
import org.apache.ambari.server.agent.RecoveryConfigComponent;
+import org.apache.ambari.server.agent.RecoveryConfigDependency;
import org.apache.ambari.server.agent.RecoveryConfigHelper;
+import org.apache.ambari.server.agent.RecoveryTopologyManager;
import
org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
@@ -72,6 +74,9 @@ public class RecoveryConfigHelperTest {
@Inject
private RecoveryConfigHelper recoveryConfigHelper;
+ @Inject
+ private RecoveryTopologyManager recoveryTopologyManager;
+
@Inject
private RepositoryVersionDAO repositoryVersionDAO;
@@ -278,6 +283,52 @@ public class RecoveryConfigHelperTest {
assertTrue(isConfigStale);
}
+ @Test
+ public void testStartTopologyContainsStatefulDependencies() throws Exception
{
+ Cluster cluster = heartbeatTestHelper.getDummyCluster();
+ RepositoryVersionEntity repositoryVersion =
helper.getOrCreateRepositoryVersion(cluster);
+ Service hdfs = cluster.addService(HDFS, repositoryVersion);
+ hdfs.addServiceComponent(NAMENODE).addServiceComponentHost(DummyHostname1);
+ hdfs.addServiceComponent("SECONDARY_NAMENODE").setRecoveryEnabled(true);
+
hdfs.getServiceComponent("SECONDARY_NAMENODE").addServiceComponentHost(DummyHostname1);
+
+
hdfs.getServiceComponent(NAMENODE).getServiceComponentHost(DummyHostname1).setState(State.STARTED);
+
hdfs.getServiceComponent(NAMENODE).getServiceComponentHost(DummyHostname1).setDesiredState(State.STARTED);
+
+ Long hostId = cluster.getHost(DummyHostname1).getHostId();
+ recoveryTopologyManager.beginAgentSession(hostId, "session-1");
+ recoveryTopologyManager.markSnapshotComplete(hostId, "session-1");
+
+ RecoveryConfig recoveryConfig =
recoveryConfigHelper.getRecoveryConfig(cluster.getClusterName(),
DummyHostname1);
+ assertTrue(recoveryConfig.isTopologyComplete());
+ assertEquals(1, recoveryConfig.getEnabledComponents().size());
+
+ RecoveryConfigComponent secondaryNameNode =
recoveryConfig.getEnabledComponents().get(0);
+ assertEquals("SECONDARY_NAMENODE", secondaryNameNode.getComponentName());
+ assertEquals(1, secondaryNameNode.getDependencies().size());
+
+ Map<String, RecoveryConfigDependency> dependencies = new HashMap<>();
+ for (RecoveryConfigDependency dependency :
secondaryNameNode.getDependencies()) {
+ dependencies.put(dependency.getComponentName(), dependency);
+ assertEquals(State.STARTED, dependency.getRequiredState());
+ assertEquals(State.STARTED, dependency.getDesiredState());
+ assertTrue(dependency.isFresh());
+ }
+ assertEquals(State.STARTED, dependencies.get(NAMENODE).getCurrentState());
+ }
+
+ @Test
+ public void testRecoveryConfigWhileServiceRemovalIsInProgress() throws
Exception {
+ Cluster cluster = heartbeatTestHelper.getDummyCluster();
+ RepositoryVersionEntity repositoryVersion =
helper.getOrCreateRepositoryVersion(cluster);
+ Service hdfs = cluster.addService(HDFS, repositoryVersion);
+
+ hdfs.delete(new DeleteHostComponentStatusMetaData());
+
+ RecoveryConfig recoveryConfig =
recoveryConfigHelper.getRecoveryConfig(cluster.getClusterName(),
DummyHostname1);
+ assertTrue(recoveryConfig.getEnabledComponents().isEmpty());
+ }
+
private Cluster getDummyCluster(Set<String> hostNames)
throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
