@Test
public void testClusterHostInfoCache() throws Exception {

  Type type = new TypeToken<Map<String, Set<String>>>() {}.getType();
  //Data for stages
  Map<String, Set<String>> clusterHostInfo1 = StageUtils.getGson().fromJson(CLUSTER_HOST_INFO, type);
  Map<String, Set<String>> clusterHostInfo2 = StageUtils.getGson().fromJson(CLUSTER_HOST_INFO_UPDATED, type);
  int stageId = 1;
  int requestId1 = 1;
  int requestId2 = 2;
  ActionQueue aq = new ActionQueue();
  Properties properties = new Properties();
  Configuration conf = new Configuration(properties);
  Clusters fsm = mock(Clusters.class);
  Cluster oneClusterMock = mock(Cluster.class);
  Service serviceObj = mock(Service.class);
  ServiceComponent scomp = mock(ServiceComponent.class);
  ServiceComponentHost sch = mock(ServiceComponentHost.class);
  UnitOfWork unitOfWork = mock(UnitOfWork.class);
  when(fsm.getCluster(anyString())).thenReturn(oneClusterMock);
  when(oneClusterMock.getService(anyString())).thenReturn(serviceObj);
  when(serviceObj.getServiceComponent(anyString())).thenReturn(scomp);
  when(scomp.getServiceComponentHost(anyString())).thenReturn(sch);
  when(serviceObj.getCluster()).thenReturn(oneClusterMock);
  Host host = mock(Host.class);
  HashMap<String, ServiceComponentHost> hosts =
          new HashMap<String, ServiceComponentHost>();
  hosts.put(hostname, sch);
  when(scomp.getServiceComponentHosts()).thenReturn(hosts);
  when(fsm.getHost(anyString())).thenReturn(host);
  when(host.getState()).thenReturn(HostState.HEALTHY);
  when(host.getHostName()).thenReturn(hostname);
  ActionDBAccessor db = mock(ActionDBAccessorImpl.class);
  HostRoleCommandDAO hostRoleCommandDAOMock = mock(HostRoleCommandDAO.class);
  Mockito.doNothing().when(hostRoleCommandDAOMock).publishTaskCreateEvent(anyListOf(HostRoleCommand.class));
  RequestEntity request = mock(RequestEntity.class);
  when(request.isExclusive()).thenReturn(false);
  when(db.getRequestEntity(anyLong())).thenReturn(request);
  Stage s1 = StageUtils.getATestStage(requestId1, stageId, hostname, CLUSTER_HOST_INFO,
    "{\"host_param\":\"param_value\"}", "{\"stage_param\":\"param_value\"}");
  Stage s2 = StageUtils.getATestStage(requestId2, stageId, hostname, CLUSTER_HOST_INFO_UPDATED,
    "{\"host_param\":\"param_value\"}", "{\"stage_param\":\"param_value\"}");
  when(db.getCommandsInProgressCount()).thenReturn(1);
  when(db.getStagesInProgress()).thenReturn(Collections.singletonList(s1));
  //Keep large number of attempts so that the task is not expired finally
  //Small action timeout to test rescheduling
  ActionScheduler scheduler = new ActionScheduler(100, 100, db, aq, fsm,
      10000, new HostsMap((String) null), unitOfWork, null, conf, entityManagerProviderMock,
      hostRoleCommandDAOMock, (HostRoleCommandFactory)null);
  scheduler.setTaskTimeoutAdjustment(false);
  List<AgentCommand> ac = waitForQueueSize(hostname, aq, 1, scheduler);
  assertTrue(ac.get(0) instanceof ExecutionCommand);
  assertEquals(String.valueOf(requestId1) + "-" + stageId, ((ExecutionCommand) (ac.get(0))).getCommandId());
  assertEquals(clusterHostInfo1, ((ExecutionCommand) (ac.get(0))).getClusterHostInfo());
  when(db.getCommandsInProgressCount()).thenReturn(1);
  when(db.getStagesInProgress()).thenReturn(Collections.singletonList(s2));
  //Verify that ActionSheduler does not return cached value of cluster host info for new requestId
  ac = waitForQueueSize(hostname, aq, 1, scheduler);
  assertTrue(ac.get(0) instanceof ExecutionCommand);
  assertEquals(String.valueOf(requestId2) + "-" + stageId, ((ExecutionCommand) (ac.get(0))).getCommandId());
  assertEquals(clusterHostInfo2, ((ExecutionCommand) (ac.get(0))).getClusterHostInfo());
}