AMBARI-24762. Ambari server continues to send request updates after all commands were completed. (#2448)

* AMBARI-24762. Ambari server continues to send request updates after all commands were completed. (mpapirkovskyy)

* AMBARI-24762. Ambari server continues to send request updates after all commands were completed. (mpapirkovskyy)

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionScheduler.java

@@ -895,6 +895,9 @@ class ActionScheduler implements Runnable {
         }
 
         updateRoleStats(status, roleStats.get(roleStr));
+        if (status == HostRoleStatus.FAILED) {
+          LOG.info("Role {} on host {} was failed", roleStr, host);
+        }
 
       }
     }

ambari-server/src/main/java/org/apache/ambari/server/events/ServiceUpdateEvent.java

@@ -21,6 +21,7 @@ package org.apache.ambari.server.events;
 import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.State;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -42,12 +43,17 @@ public class ServiceUpdateEvent extends STOMPEvent {
   @JsonProperty("state")
   private State state;
 
-  public ServiceUpdateEvent(String clusterName, MaintenanceState maintenanceState, String serviceName, State state) {
+  @JsonIgnore
+  private boolean stateChanged = false;
+
+  public ServiceUpdateEvent(String clusterName, MaintenanceState maintenanceState, String serviceName, State state,
+                            boolean stateChanged) {
     super(Type.SERVICE);
     this.clusterName = clusterName;
     this.maintenanceState = maintenanceState;
     this.serviceName = serviceName;
     this.state = state;
+    this.stateChanged = stateChanged;
   }
 
   public String getClusterName() {
@@ -82,6 +88,14 @@ public class ServiceUpdateEvent extends STOMPEvent {
     this.state = state;
   }
 
+  public boolean isStateChanged() {
+    return stateChanged;
+  }
+
+  public void setStateChanged(boolean stateChanged) {
+    this.stateChanged = stateChanged;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;

ambari-server/src/main/java/org/apache/ambari/server/events/listeners/services/ServiceUpdateListener.java

@@ -24,18 +24,14 @@ import java.util.Set;
 
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.EagerSingleton;
-import org.apache.ambari.server.controller.utilities.ServiceCalculatedStateFactory;
-import org.apache.ambari.server.controller.utilities.state.ServiceCalculatedState;
 import org.apache.ambari.server.events.HostComponentUpdate;
 import org.apache.ambari.server.events.HostComponentsUpdateEvent;
 import org.apache.ambari.server.events.MaintenanceModeEvent;
 import org.apache.ambari.server.events.ServiceUpdateEvent;
 import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
 import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
-import org.apache.ambari.server.orm.dao.ServiceDesiredStateDAO;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.MaintenanceState;
-import org.apache.ambari.server.state.State;
 
 import com.google.common.eventbus.Subscribe;
 import com.google.inject.Inject;
@@ -45,13 +41,9 @@ import com.google.inject.Singleton;
 @Singleton
 @EagerSingleton
 public class ServiceUpdateListener {
-  private Map<Long, Map<String, State>> states = new HashMap<>();
 
   private STOMPUpdatePublisher STOMPUpdatePublisher;
 
-  @Inject
-  private ServiceDesiredStateDAO serviceDesiredStateDAO;
-
   @Inject
   private Provider<Clusters> m_clusters;
 
@@ -75,15 +67,8 @@ public class ServiceUpdateListener {
       Long clusterId = clusterServices.getKey();
       String clusterName = m_clusters.get().getClusterById(clusterId).getClusterName();
       for (String serviceName : clusterServices.getValue()) {
-        ServiceCalculatedState serviceCalculatedState = ServiceCalculatedStateFactory.getServiceStateProvider(serviceName);
-        State serviceState = serviceCalculatedState.getState(clusterName, serviceName);
-
-        // retrieve state from cache
-        if (states.containsKey(clusterId) && states.get(clusterId).containsKey(serviceName) && states.get(clusterId).get(serviceName).equals(serviceState)) {
-          continue;
-        }
-        states.computeIfAbsent(clusterId, c -> new HashMap<>()).put(serviceName, serviceState);
-        STOMPUpdatePublisher.publish(new ServiceUpdateEvent(clusterName, null, serviceName, serviceState));
+        STOMPUpdatePublisher.publish(new ServiceUpdateEvent(clusterName, null, serviceName, null,
+            true));
       }
     }
   }
@@ -99,6 +84,7 @@ public class ServiceUpdateListener {
 
     MaintenanceState maintenanceState = event.getMaintenanceState();
 
-    STOMPUpdatePublisher.publish(new ServiceUpdateEvent(clusterName, maintenanceState, serviceName, null));
+    STOMPUpdatePublisher.publish(new ServiceUpdateEvent(clusterName, maintenanceState, serviceName, null,
+        false));
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/events/publishers/BufferedUpdateEventPublisher.java

@@ -24,8 +24,6 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
 
 import com.google.common.eventbus.EventBus;
 import com.google.inject.Singleton;
@@ -34,31 +32,25 @@ import com.google.inject.Singleton;
 public abstract class BufferedUpdateEventPublisher<T> {
 
   private static final long TIMEOUT = 1000L;
-  private final AtomicLong previousTime = new AtomicLong(0);
-  private final AtomicBoolean collecting = new AtomicBoolean(false);
   private final ConcurrentLinkedQueue<T> buffer = new ConcurrentLinkedQueue<>();
-  private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+
+  private ScheduledExecutorService scheduledExecutorService;
 
   public void publish(T event, EventBus m_eventBus) {
-    long eventTime = System.currentTimeMillis();
-    if ((eventTime - previousTime.get() <= TIMEOUT) && !collecting.get()) {
-      buffer.add(event);
-      collecting.set(true);
-      scheduledExecutorService.schedule(getScheduledPublisher(m_eventBus),
-          TIMEOUT, TimeUnit.MILLISECONDS);
-    } else if (collecting.get()) {
-      buffer.add(event);
-    } else {
-      //TODO add logging and metrics posting
-      previousTime.set(eventTime);
-      m_eventBus.post(event);
+    if (scheduledExecutorService == null) {
+      scheduledExecutorService =
+          Executors.newScheduledThreadPool(1);
+      scheduledExecutorService
+          .scheduleWithFixedDelay(getScheduledPublisher(m_eventBus), TIMEOUT, TIMEOUT, TimeUnit.MILLISECONDS);
     }
+    buffer.add(event);
   }
 
-  protected abstract Runnable getScheduledPublisher(EventBus m_eventBus);
+  protected MergingRunnable getScheduledPublisher(EventBus m_eventBus) {
+    return new MergingRunnable(m_eventBus);
+  }
 
   protected List<T> retrieveBuffer() {
-    resetCollecting();
     List<T> bufferContent = new ArrayList<>();
     while (!buffer.isEmpty()) {
       bufferContent.add(buffer.poll());
@@ -66,8 +58,23 @@ public abstract class BufferedUpdateEventPublisher<T> {
     return bufferContent;
   }
 
-  protected void resetCollecting() {
-    previousTime.set(System.currentTimeMillis());
-    collecting.set(false);
+  public abstract void mergeBufferAndPost(List<T> events, EventBus m_eventBus);
+
+  private class MergingRunnable implements Runnable {
+
+    private final EventBus m_eventBus;
+
+    public MergingRunnable(EventBus m_eventBus) {
+      this.m_eventBus = m_eventBus;
+    }
+
+    @Override
+    public final void run() {
+      List<T> events = retrieveBuffer();
+      if (events.isEmpty()) {
+        return;
+      }
+      mergeBufferAndPost(events, m_eventBus);
+    }
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/events/publishers/HostComponentUpdateEventPublisher.java

@@ -31,30 +31,12 @@ import com.google.inject.Singleton;
 public class HostComponentUpdateEventPublisher extends BufferedUpdateEventPublisher<HostComponentsUpdateEvent> {
 
   @Override
-  protected Runnable getScheduledPublisher(EventBus m_eventBus) {
-    return new HostComponentsEventRunnable(m_eventBus);
-  }
-
-  private class HostComponentsEventRunnable implements Runnable {
-
-    private final EventBus eventBus;
-
-    public HostComponentsEventRunnable(EventBus eventBus) {
-      this.eventBus = eventBus;
-    }
-
-    @Override
-    public void run() {
-      List<HostComponentsUpdateEvent> hostComponentUpdateEvents = retrieveBuffer();
-      if (hostComponentUpdateEvents.isEmpty()) {
-        return;
-      }
-      List<HostComponentUpdate> hostComponentUpdates = hostComponentUpdateEvents.stream().flatMap(
-          u -> u.getHostComponentUpdates().stream()).collect(Collectors.toList());
+  public void mergeBufferAndPost(List<HostComponentsUpdateEvent> events, EventBus m_eventBus) {
+    List<HostComponentUpdate> hostComponentUpdates = events.stream().flatMap(
+        u -> u.getHostComponentUpdates().stream()).collect(Collectors.toList());
 
-      HostComponentsUpdateEvent resultEvents = new HostComponentsUpdateEvent(hostComponentUpdates);
-      //TODO add logging and metrics posting
-      eventBus.post(resultEvents);
-    }
+    HostComponentsUpdateEvent resultEvents = new HostComponentsUpdateEvent(hostComponentUpdates);
+    //TODO add logging and metrics posting
+    m_eventBus.post(resultEvents);
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/events/publishers/RequestUpdateEventPublisher.java

@@ -18,9 +18,9 @@
 
 package org.apache.ambari.server.events.publishers;
 
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 import org.apache.ambari.server.controller.internal.CalculatedStatus;
 import org.apache.ambari.server.events.RequestUpdateEvent;
@@ -35,11 +35,7 @@ import com.google.inject.Inject;
 import com.google.inject.Singleton;
 
 @Singleton
-public class RequestUpdateEventPublisher {
-
-  private final Long TIMEOUT = 1000L;
-  private ConcurrentHashMap<Long, Long> previousTime = new ConcurrentHashMap<>();
-  private ConcurrentHashMap<Long, RequestUpdateEvent> buffer = new ConcurrentHashMap<>();
+public class RequestUpdateEventPublisher extends BufferedUpdateEventPublisher<RequestUpdateEvent> {
 
   @Inject
   private HostRoleCommandDAO hostRoleCommandDAO;
@@ -53,27 +49,25 @@ public class RequestUpdateEventPublisher {
   @Inject
   private ClusterDAO clusterDAO;
 
-  public void publish(RequestUpdateEvent event, EventBus m_eventBus) {
-    Long eventTime = System.currentTimeMillis();
-    Long requestId = event.getRequestId();
-    if (!previousTime.containsKey(requestId)) {
-      previousTime.put(requestId, 0L);
+  @Override
+  public void mergeBufferAndPost(List<RequestUpdateEvent> events, EventBus m_eventBus) {
+    Map<Long, RequestUpdateEvent> filteredRequests = new HashMap<>();
+    for (RequestUpdateEvent event : events) {
+      Long requestId = event.getRequestId();
+      if (filteredRequests.containsKey(requestId)) {
+        RequestUpdateEvent filteredRequest = filteredRequests.get(requestId);
+        filteredRequest.setEndTime(event.getEndTime());
+        filteredRequest.setRequestStatus(event.getRequestStatus());
+        filteredRequest.setRequestContext(event.getRequestContext());
+        filteredRequest.getHostRoleCommands().removeAll(event.getHostRoleCommands());
+        filteredRequest.getHostRoleCommands().addAll(event.getHostRoleCommands());
+      } else {
+        filteredRequests.put(requestId, event);
+      }
     }
-    if (eventTime - previousTime.get(requestId) <= TIMEOUT && !buffer.containsKey(requestId)) {
-      buffer.put(event.getRequestId(), event);
-      Executors.newScheduledThreadPool(1).schedule(new RequestEventRunnable(requestId, m_eventBus),
-          TIMEOUT, TimeUnit.MILLISECONDS);
-    } else if (buffer.containsKey(requestId)) {
-      //merge available buffer content with arrived
-      buffer.get(requestId).setEndTime(event.getEndTime());
-      buffer.get(requestId).setRequestStatus(event.getRequestStatus());
-      buffer.get(requestId).setRequestContext(event.getRequestContext());
-      buffer.get(requestId).getHostRoleCommands().removeAll(event.getHostRoleCommands());
-      buffer.get(requestId).getHostRoleCommands().addAll(event.getHostRoleCommands());
-    } else {
-      previousTime.put(requestId, eventTime);
-      //TODO add logging and metrics posting
-      m_eventBus.post(fillRequest(event));
+    for (RequestUpdateEvent requestUpdateEvent : filteredRequests.values()) {
+      RequestUpdateEvent filled = fillRequest(requestUpdateEvent);
+      m_eventBus.post(filled);
     }
   }
 
@@ -94,24 +88,4 @@ public class RequestUpdateEventPublisher {
     }
     return event;
   }
-
-  private class RequestEventRunnable implements Runnable {
-
-    private final long requestId;
-    private final EventBus eventBus;
-
-    public RequestEventRunnable(long requestId, EventBus eventBus) {
-      this.requestId = requestId;
-      this.eventBus = eventBus;
-    }
-
-    @Override
-    public void run() {
-      RequestUpdateEvent resultEvent = buffer.get(requestId);
-      //TODO add logging and metrics posting
-      eventBus.post(fillRequest(resultEvent));
-      buffer.remove(requestId);
-      previousTime.remove(requestId);
-    }
-  }
 }

ambari-server/src/main/java/org/apache/ambari/server/events/publishers/ServiceUpdateEventPublisher.java

@@ -19,52 +19,62 @@
 package org.apache.ambari.server.events.publishers;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
+import org.apache.ambari.server.controller.utilities.ServiceCalculatedStateFactory;
+import org.apache.ambari.server.controller.utilities.state.ServiceCalculatedState;
 import org.apache.ambari.server.events.ServiceUpdateEvent;
+import org.apache.ambari.server.state.State;
 
 import com.google.common.eventbus.EventBus;
 import com.google.inject.Singleton;
 
 @Singleton
 public class ServiceUpdateEventPublisher extends BufferedUpdateEventPublisher<ServiceUpdateEvent> {
+  private Map<String, Map<String, State>> states = new HashMap<>();
 
-  @Override
-  protected Runnable getScheduledPublisher(EventBus m_eventBus) {
-    return new ServiceEventRunnable(m_eventBus);
-  }
-
-  private class ServiceEventRunnable implements Runnable {
 
-    private final EventBus eventBus;
-
-    public ServiceEventRunnable(EventBus eventBus) {
-      this.eventBus = eventBus;
+  @Override
+  public void mergeBufferAndPost(List<ServiceUpdateEvent> events, EventBus eventBus) {
+    List<ServiceUpdateEvent> filtered = new ArrayList<>();
+    for (ServiceUpdateEvent event : events) {
+      int pos = filtered.indexOf(event);
+      if (pos != -1) {
+        if (event.isStateChanged()) {
+          filtered.get(pos).setStateChanged(true);
+        }
+        if (event.getMaintenanceState() != null) {
+          filtered.get(pos).setMaintenanceState(event.getMaintenanceState());
+        }
+      } else {
+        filtered.add(event);
+      }
     }
+    for (ServiceUpdateEvent serviceUpdateEvent : filtered) {
+      // calc state
+      if (serviceUpdateEvent.isStateChanged()) {
+        ServiceCalculatedState serviceCalculatedState =
+            ServiceCalculatedStateFactory.getServiceStateProvider(serviceUpdateEvent.getServiceName());
+        State serviceState =
+            serviceCalculatedState.getState(serviceUpdateEvent.getClusterName(), serviceUpdateEvent.getServiceName());
 
-    @Override
-    public void run() {
-      List<ServiceUpdateEvent> serviceUpdates = retrieveBuffer();
-      if (serviceUpdates.isEmpty()) {
-        return;
-      }
-      List<ServiceUpdateEvent> filtered = new ArrayList<>();
-      for (ServiceUpdateEvent event : serviceUpdates) {
-        int pos = filtered.indexOf(event);
-        if (pos != -1) {
-          if (event.getState() != null) {
-            filtered.get(pos).setState(event.getState());
-          }
-          if (event.getMaintenanceState() != null) {
-            filtered.get(pos).setMaintenanceState(event.getMaintenanceState());
-          }
-        } else {
-          filtered.add(event);
+        String serviceName = serviceUpdateEvent.getServiceName();
+        String clusterName = serviceUpdateEvent.getClusterName();
+
+        // retrieve state from cache
+        // don't send update when state was not changed and update doesn't have maintenance info
+        if (states.containsKey(clusterName) && states.get(clusterName).containsKey(serviceName)
+            && states.get(clusterName).get(serviceName).equals(serviceState)
+            && serviceUpdateEvent.getMaintenanceState() == null) {
+          continue;
         }
+        states.computeIfAbsent(clusterName, c -> new HashMap<>()).put(serviceName, serviceState);
+        serviceUpdateEvent.setState(serviceState);
       }
-      for (ServiceUpdateEvent serviceUpdateEvent : serviceUpdates) {
-        eventBus.post(serviceUpdateEvent);
-      }
+
+      eventBus.post(serviceUpdateEvent);
     }
   }
 }