samza-yarn/src/main/java/org/apache/samza/webapp/YarnContainerHeartbeatServlet.java [49:89]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class YarnContainerHeartbeatServlet extends HttpServlet {

  private static final Logger LOG = LoggerFactory.getLogger(YarnContainerHeartbeatServlet.class);
  private static final String APPLICATION_JSON = "application/json";
  private static final String GROUP = SamzaAppMasterMetrics.class.getName();
  private final Counter heartbeatsExpiredCount;

  private YarnAppState yarnAppState;
  private ObjectMapper mapper;

  public YarnContainerHeartbeatServlet(YarnAppState yarnAppState, ReadableMetricsRegistry registry) {
    this.yarnAppState = yarnAppState;
    this.mapper = new ObjectMapper();
    this.heartbeatsExpiredCount = registry.newCounter(GROUP, "heartbeats-expired");
  }

  @Override
  protected void doGet(HttpServletRequest req, HttpServletResponse resp)
      throws ServletException, IOException {
    ContainerId yarnContainerId;
    PrintWriter printWriter = resp.getWriter();
    String containerIdParam = req.getParameter(CoordinationConstants.YARN_EXECUTION_ENVIRONMENT_CONTAINER_ID);
    ContainerHeartbeatResponse response;
    resp.setContentType(APPLICATION_JSON);
    boolean alive = false;
    try {
      yarnContainerId = ContainerId.fromString(containerIdParam);
      for (YarnContainer yarnContainer : yarnAppState.runningProcessors.values()) {
        if (yarnContainer.id().compareTo(yarnContainerId) == 0) {
          alive = true;
          break;
        }
      }
      if (!alive) {
        heartbeatsExpiredCount.inc();
      }
      response = new ContainerHeartbeatResponse(alive);
      printWriter.write(mapper.writeValueAsString(response));
    } catch (IllegalArgumentException e) {
      LOG.error("Container ID {} passed is invalid", containerIdParam);
      resp.sendError(HttpServletResponse.SC_BAD_REQUEST, e.getMessage());
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



samza-yarn3/src/main/java/org/apache/samza/webapp/YarnContainerHeartbeatServlet.java [49:89]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
public class YarnContainerHeartbeatServlet extends HttpServlet {

  private static final Logger LOG = LoggerFactory.getLogger(YarnContainerHeartbeatServlet.class);
  private static final String APPLICATION_JSON = "application/json";
  private static final String GROUP = SamzaAppMasterMetrics.class.getName();
  private final Counter heartbeatsExpiredCount;

  private YarnAppState yarnAppState;
  private ObjectMapper mapper;

  public YarnContainerHeartbeatServlet(YarnAppState yarnAppState, ReadableMetricsRegistry registry) {
    this.yarnAppState = yarnAppState;
    this.mapper = new ObjectMapper();
    this.heartbeatsExpiredCount = registry.newCounter(GROUP, "heartbeats-expired");
  }

  @Override
  protected void doGet(HttpServletRequest req, HttpServletResponse resp)
      throws ServletException, IOException {
    ContainerId yarnContainerId;
    PrintWriter printWriter = resp.getWriter();
    String containerIdParam = req.getParameter(CoordinationConstants.YARN_EXECUTION_ENVIRONMENT_CONTAINER_ID);
    ContainerHeartbeatResponse response;
    resp.setContentType(APPLICATION_JSON);
    boolean alive = false;
    try {
      yarnContainerId = ContainerId.fromString(containerIdParam);
      for (YarnContainer yarnContainer : yarnAppState.runningProcessors.values()) {
        if (yarnContainer.id().compareTo(yarnContainerId) == 0) {
          alive = true;
          break;
        }
      }
      if (!alive) {
        heartbeatsExpiredCount.inc();
      }
      response = new ContainerHeartbeatResponse(alive);
      printWriter.write(mapper.writeValueAsString(response));
    } catch (IllegalArgumentException e) {
      LOG.error("Container ID {} passed is invalid", containerIdParam);
      resp.sendError(HttpServletResponse.SC_BAD_REQUEST, e.getMessage());
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -