StreamingContainerManager implements PlanContext { @SuppressWarnings("StatementWithEmptyBody") public ContainerHeartbeatResponse processHeartbeat(ContainerHeartbeat heartbeat) { long currentTimeMillis = clock.getTime(); final StreamingContainerAgent sca = this.containers.get(heartbeat.getContainerId()); if (sca == null || sca.container.getState() == PTContainer.State.KILLED) { LOG.error("Unknown container {}", heartbeat.getContainerId()); ContainerHeartbeatResponse response = new ContainerHeartbeatResponse(); response.shutdown = ShutdownType.ABORT; return response; } if (sca.container.getState() == PTContainer.State.ALLOCATED) { if (sca.container.bufferServerAddress == null && heartbeat.bufferServerHost != null) { sca.container.bufferServerAddress = InetSocketAddress.createUnresolved(heartbeat.bufferServerHost, heartbeat.bufferServerPort); LOG.info("Container {} buffer server: {}", sca.container.getExternalId(), sca.container.bufferServerAddress); } final long containerStartTime = System.currentTimeMillis(); sca.container.setState(PTContainer.State.ACTIVE); sca.container.setStartedTime(containerStartTime); sca.container.setFinishedTime(-1); sca.jvmName = heartbeat.jvmName; poolExecutor.submit(new Runnable() { @Override public void run() { try { containerFile.append(sca.getContainerInfo()); } catch (IOException ex) { LOG.warn("Cannot write to container file"); } for (PTOperator ptOp : sca.container.getOperators()) { try { JSONObject operatorInfo = new JSONObject(); operatorInfo.put("name", ptOp.getName()); operatorInfo.put("id", ptOp.getId()); operatorInfo.put("container", sca.container.getExternalId()); operatorInfo.put("startTime", containerStartTime); operatorFile.append(operatorInfo); } catch (IOException | JSONException ex) { LOG.warn("Cannot write to operator file: ", ex); } } } }); } sca.containerStackTrace = heartbeat.stackTrace; if (heartbeat.restartRequested) { LOG.error("Container {} restart request", sca.container.getExternalId()); containerStopRequests.put(sca.container.getExternalId(), sca.container.getExternalId()); } sca.memoryMBFree = heartbeat.memoryMBFree; sca.gcCollectionCount = heartbeat.gcCollectionCount; sca.gcCollectionTime = heartbeat.gcCollectionTime; sca.undeployOpers.clear(); sca.deployOpers.clear(); if (!this.deployChangeInProgress.get()) { sca.deployCnt = this.deployChangeCnt; } Set<Integer> reportedOperators = Sets.newHashSetWithExpectedSize(sca.container.getOperators().size()); for (OperatorHeartbeat shb : heartbeat.getContainerStats().operators) { long maxEndWindowTimestamp = 0; reportedOperators.add(shb.nodeId); PTOperator oper = this.plan.getAllOperators().get(shb.getNodeId()); if (oper == null) { LOG.info("Heartbeat for unknown operator {} (container {})", shb.getNodeId(), heartbeat.getContainerId()); sca.undeployOpers.add(shb.nodeId); continue; } if (shb.requestResponse != null) { for (StatsListener.OperatorResponse obj : shb.requestResponse) { if (obj instanceof OperatorResponse) { commandResponse.put((Long)obj.getResponseId(), obj.getResponse()); LOG.debug(" Got back the response {} for the request {}", obj, obj.getResponseId()); } else { oper.stats.responses.add(obj); } } } if (!(oper.getState() == PTOperator.State.ACTIVE && shb.getState() == OperatorHeartbeat.DeployState.ACTIVE)) { processOperatorDeployStatus(oper, shb, sca); } oper.stats.lastHeartbeat = shb; List<ContainerStats.OperatorStats> statsList = shb.getOperatorStatsContainer(); if (!statsList.isEmpty()) { long tuplesProcessed = 0; long tuplesEmitted = 0; long totalCpuTimeUsed = 0; int statCount = 0; long maxDequeueTimestamp = -1; oper.stats.recordingId = null; final OperatorStatus status = oper.stats; status.statsRevs.checkout(); for (Map.Entry<String, PortStatus> entry : status.inputPortStatusList.entrySet()) { entry.getValue().recordingId = null; } for (Map.Entry<String, PortStatus> entry : status.outputPortStatusList.entrySet()) { entry.getValue().recordingId = null; } for (ContainerStats.OperatorStats stats : statsList) { if (stats == null) { LOG.warn("Operator {} statistics list contains null element", shb.getNodeId()); continue; } if (stats.checkpoint instanceof Checkpoint) { if (oper.getRecentCheckpoint() == null || oper.getRecentCheckpoint().windowId < stats.checkpoint.getWindowId()) { addCheckpoint(oper, (Checkpoint)stats.checkpoint); if (stats.checkpointStats != null) { status.checkpointStats = stats.checkpointStats; status.checkpointTimeMA.add(stats.checkpointStats.checkpointTime); } oper.failureCount = 0; } } oper.stats.recordingId = stats.recordingId; EndWindowStats endWindowStats = new EndWindowStats(); Collection<ContainerStats.OperatorStats.PortStats> ports = stats.inputPorts; if (ports != null) { Set<String> currentInputPortSet = Sets.newHashSetWithExpectedSize(ports.size()); for (ContainerStats.OperatorStats.PortStats s : ports) { currentInputPortSet.add(s.id); PortStatus ps = status.inputPortStatusList.get(s.id); if (ps == null) { ps = status.new PortStatus(); ps.portName = s.id; status.inputPortStatusList.put(s.id, ps); } ps.totalTuples += s.tupleCount; ps.recordingId = s.recordingId; tuplesProcessed += s.tupleCount; endWindowStats.dequeueTimestamps.put(s.id, s.endWindowTimestamp); Pair<Integer, String> operatorPortName = new Pair<>(oper.getId(), s.id); Long lastEndWindowTimestamp = operatorPortLastEndWindowTimestamps.get(operatorPortName); if (lastEndWindowTimestamp == null) { lastEndWindowTimestamp = lastStatsTimestamp; } long portElapsedMillis = Math.max(s.endWindowTimestamp - lastEndWindowTimestamp, 0); ps.tuplesPMSMA.add(s.tupleCount, portElapsedMillis); ps.bufferServerBytesPMSMA.add(s.bufferServerBytes, portElapsedMillis); ps.queueSizeMA.add(s.queueSize); operatorPortLastEndWindowTimestamps.put(operatorPortName, s.endWindowTimestamp); if (maxEndWindowTimestamp < s.endWindowTimestamp) { maxEndWindowTimestamp = s.endWindowTimestamp; } if (s.endWindowTimestamp > maxDequeueTimestamp) { maxDequeueTimestamp = s.endWindowTimestamp; } } Iterator<Map.Entry<String, PortStatus>> it = status.inputPortStatusList.entrySet().iterator(); while (it.hasNext()) { Map.Entry<String, PortStatus> entry = it.next(); if (!currentInputPortSet.contains(entry.getKey())) { it.remove(); } } } ports = stats.outputPorts; if (ports != null) { Set<String> currentOutputPortSet = Sets.newHashSetWithExpectedSize(ports.size()); for (ContainerStats.OperatorStats.PortStats s : ports) { currentOutputPortSet.add(s.id); PortStatus ps = status.outputPortStatusList.get(s.id); if (ps == null) { ps = status.new PortStatus(); ps.portName = s.id; status.outputPortStatusList.put(s.id, ps); } ps.totalTuples += s.tupleCount; ps.recordingId = s.recordingId; tuplesEmitted += s.tupleCount; Pair<Integer, String> operatorPortName = new Pair<>(oper.getId(), s.id); Long lastEndWindowTimestamp = operatorPortLastEndWindowTimestamps.get(operatorPortName); if (lastEndWindowTimestamp == null) { lastEndWindowTimestamp = lastStatsTimestamp; } long portElapsedMillis = Math.max(s.endWindowTimestamp - lastEndWindowTimestamp, 0); ps.tuplesPMSMA.add(s.tupleCount, portElapsedMillis); ps.bufferServerBytesPMSMA.add(s.bufferServerBytes, portElapsedMillis); operatorPortLastEndWindowTimestamps.put(operatorPortName, s.endWindowTimestamp); if (maxEndWindowTimestamp < s.endWindowTimestamp) { maxEndWindowTimestamp = s.endWindowTimestamp; } } if (ports.size() > 0) { endWindowStats.emitTimestamp = ports.iterator().next().endWindowTimestamp; } Iterator<Map.Entry<String, PortStatus>> it = status.outputPortStatusList.entrySet().iterator(); while (it.hasNext()) { Map.Entry<String, PortStatus> entry = it.next(); if (!currentOutputPortSet.contains(entry.getKey())) { it.remove(); } } } if (endWindowStats.emitTimestamp < 0) { endWindowStats.emitTimestamp = maxDequeueTimestamp; } if (status.currentWindowId.get() != stats.windowId) { status.lastWindowIdChangeTms = currentTimeMillis; status.currentWindowId.set(stats.windowId); } totalCpuTimeUsed += stats.cpuTimeUsed; statCount++; if (oper.getOperatorMeta().getValue(OperatorContext.COUNTERS_AGGREGATOR) != null) { endWindowStats.counters = stats.counters; } if (oper.getOperatorMeta().getMetricAggregatorMeta() != null && oper.getOperatorMeta().getMetricAggregatorMeta().getAggregator() != null) { endWindowStats.metrics = stats.metrics; } if (stats.windowId > currentEndWindowStatsWindowId) { Map<Integer, EndWindowStats> endWindowStatsMap = endWindowStatsOperatorMap.get(stats.windowId); if (endWindowStatsMap == null) { endWindowStatsMap = new ConcurrentSkipListMap<>(); Map<Integer, EndWindowStats> endWindowStatsMapPrevious = endWindowStatsOperatorMap.putIfAbsent(stats.windowId, endWindowStatsMap); if (endWindowStatsMapPrevious != null) { endWindowStatsMap = endWindowStatsMapPrevious; } } endWindowStatsMap.put(shb.getNodeId(), endWindowStats); Set<Integer> allCurrentOperators = plan.getAllOperators().keySet(); int numOperators = plan.getAllOperators().size(); if (allCurrentOperators.containsAll(endWindowStatsMap.keySet()) && endWindowStatsMap.size() == numOperators) { completeEndWindowStatsWindowId = stats.windowId; } } } status.totalTuplesProcessed.add(tuplesProcessed); status.totalTuplesEmitted.add(tuplesEmitted); OperatorMeta logicalOperator = oper.getOperatorMeta(); LogicalOperatorStatus logicalStatus = logicalOperator.getStatus(); if (!oper.isUnifier()) { logicalStatus.totalTuplesProcessed += tuplesProcessed; logicalStatus.totalTuplesEmitted += tuplesEmitted; } long lastMaxEndWindowTimestamp = operatorLastEndWindowTimestamps.containsKey(oper.getId()) ? operatorLastEndWindowTimestamps.get(oper.getId()) : lastStatsTimestamp; if (maxEndWindowTimestamp >= lastMaxEndWindowTimestamp) { double tuplesProcessedPMSMA = 0.0; double tuplesEmittedPMSMA = 0.0; if (statCount != 0) { status.cpuNanosPMSMA.add(totalCpuTimeUsed, maxEndWindowTimestamp - lastMaxEndWindowTimestamp); } for (PortStatus ps : status.inputPortStatusList.values()) { tuplesProcessedPMSMA += ps.tuplesPMSMA.getAvg(); } for (PortStatus ps : status.outputPortStatusList.values()) { tuplesEmittedPMSMA += ps.tuplesPMSMA.getAvg(); } status.tuplesProcessedPSMA.set(Math.round(tuplesProcessedPMSMA * 1000)); status.tuplesEmittedPSMA.set(Math.round(tuplesEmittedPMSMA * 1000)); } else { } operatorLastEndWindowTimestamps.put(oper.getId(), maxEndWindowTimestamp); status.listenerStats.add(statsList); this.reportStats.put(oper, oper); status.statsRevs.commit(); } if (lastStatsTimestamp < maxEndWindowTimestamp) { lastStatsTimestamp = maxEndWindowTimestamp; } } sca.lastHeartbeatMillis = currentTimeMillis; for (PTOperator oper : sca.container.getOperators()) { if (!reportedOperators.contains(oper.getId())) { processOperatorDeployStatus(oper, null, sca); } } ContainerHeartbeatResponse rsp = getHeartbeatResponse(sca); if (heartbeat.getContainerStats().operators.isEmpty() && isApplicationIdle()) { LOG.info("requesting idle shutdown for container {}", heartbeat.getContainerId()); rsp.shutdown = ShutdownType.ABORT; } else { if (sca.isShutdownRequested()) { LOG.info("requesting shutdown for container {}", heartbeat.getContainerId()); rsp.shutdown = sca.shutdownRequest; } } List<StramToNodeRequest> requests = rsp.nodeRequests != null ? rsp.nodeRequests : new ArrayList<StramToNodeRequest>(); ConcurrentLinkedQueue<StramToNodeRequest> operatorRequests = sca.getOperatorRequests(); while (true) { StramToNodeRequest r = operatorRequests.poll(); if (r == null) { break; } requests.add(r); } rsp.nodeRequests = requests; rsp.committedWindowId = committedWindowId; rsp.stackTraceRequired = sca.stackTraceRequested; sca.stackTraceRequested = false; apexPluginDispatcher.dispatch(new DAGExecutionEvent.HeartbeatExecutionEvent(heartbeat)); return rsp; } StreamingContainerManager(LogicalPlan dag, Clock clock);  StreamingContainerManager(LogicalPlan dag);  StreamingContainerManager(LogicalPlan dag, boolean enableEventRecording, Clock clock); private  StreamingContainerManager(CheckpointState checkpointedState, boolean enableEventRecording); Journal getJournal(); final ContainerInfo getAppMasterContainerInfo(); void updateRPCLatency(String containerId, long latency); void teardown(); void subscribeToEvents(Object listener); PhysicalPlan getPhysicalPlan(); long getCommittedWindowId(); boolean isGatewayConnected(); SharedPubSubWebSocketClient getWsClient(); List<AppDataSource> getAppDataSources(); Map<String, Map<String, Object>> getLatestLogicalMetrics(); void monitorHeartbeat(boolean waitForRecovery); Queue<Pair<Long, Map<String, Object>>> getWindowMetrics(String operatorName); int processEvents(); void scheduleContainerRestart(String containerId); void removeContainerAgent(String containerId); Collection<ContainerInfo> getCompletedContainerInfo(); StreamingContainerAgent assignContainer(ContainerResource resource, InetSocketAddress bufferServerAddr); StreamingContainerAgent getContainerAgent(String containerId); Collection<StreamingContainerAgent> getContainerAgents(); @SuppressWarnings("StatementWithEmptyBody") ContainerHeartbeatResponse processHeartbeat(ContainerHeartbeat heartbeat); long updateOperatorLatency(PTOperator oper, UpdateOperatorLatencyContext ctx); void updateRecoveryCheckpoints(PTOperator operator, UpdateCheckpointsContext ctx, boolean recovery); long windowIdToMillis(long windowId); long getWindowStartMillis(); void shutdownAllContainers(ShutdownType type, String message); void deployAfterRestart(); @Override void deploy(Set<PTContainer> releaseContainers, Collection<PTOperator> undeploy, Set<PTContainer> startContainers, Collection<PTOperator> deploy); @Override void recordEventAsync(StramEvent ev); @Override void dispatch(Runnable r); OperatorInfo getOperatorInfo(int operatorId); List<OperatorInfo> getOperatorInfoList(); LogicalOperatorInfo getLogicalOperatorInfo(String operatorName); ModuleMeta getModuleMeta(String moduleName); List<LogicalOperatorInfo> getLogicalOperatorInfoList(); OperatorAggregationInfo getOperatorAggregationInfo(String operatorName); static long toWsWindowId(long windowId); List<StreamInfo> getStreamInfoList(); void startRecording(String id, int operId, String portName, long numWindows); void stopRecording(int operId, String portName); void syncStats(); void syncEvents(); void stopContainer(String containerId); Recoverable getSetOperatorProperty(String operatorName, String propertyName, String propertyValue); Recoverable getSetPhysicalOperatorProperty(int operatorId, String propertyName, String propertyValue); void setOperatorProperty(String operatorName, String propertyName, String propertyValue); void setPhysicalOperatorProperty(int operatorId, String propertyName, String propertyValue); @Override void addOperatorRequest(PTOperator oper, StramToNodeRequest request); void setLoggersLevel(Map<String, String> changedLoggers); FutureTask<Object> getPhysicalOperatorProperty(int operatorId, String propertyName, long waitTime); Attribute.AttributeMap getApplicationAttributes(); Attribute.AttributeMap getOperatorAttributes(String operatorId); Attribute.AttributeMap getPortAttributes(String operatorId, String portName); LogicalPlan getLogicalPlan(); FutureTask<Object> logicalPlanModification(List<LogicalPlanRequest> requests); CriticalPathInfo getCriticalPathInfo(); @Override void writeJournal(Recoverable operation); static StreamingContainerManager getInstance(RecoveryHandler rh, LogicalPlan dag, boolean enableEventRecording); void setApexPluginDispatcher(ApexPluginDispatcher manager); static final String BUILTIN_APPDATA_URL; static final String CONTAINERS_INFO_FILENAME_FORMAT; static final String OPERATORS_INFO_FILENAME_FORMAT; static final String APP_META_FILENAME; static final String APP_META_KEY_ATTRIBUTES; static final String APP_META_KEY_METRICS; static final String EMBEDDABLE_QUERY_NAME_SUFFIX; static final Recoverable SET_OPERATOR_PROPERTY; static final Recoverable SET_PHYSICAL_OPERATOR_PROPERTY; static final int METRIC_QUEUE_SIZE; transient ApexPluginDispatcher apexPluginDispatcher; }
@Test public void testProcessHeartbeat() throws Exception { TestGeneratorInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.class); dag.setOperatorAttribute(o1, OperatorContext.STATS_LISTENERS, Arrays.asList(new StatsListener[]{new PartitioningTest.PartitionLoadWatch()})); dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent()); StreamingContainerManager scm = new StreamingContainerManager(dag); PhysicalPlan plan = scm.getPhysicalPlan(); Assert.assertEquals("number required containers", 1, plan.getContainers().size()); PTOperator o1p1 = plan.getOperators(dag.getMeta(o1)).get(0); String containerId = "container1"; StreamingContainerAgent sca = scm.assignContainer(new ContainerResource(0, containerId, "localhost", 512, 0, null), InetSocketAddress.createUnresolved("localhost", 0)); Assert.assertNotNull(sca); Assert.assertEquals(PTContainer.State.ALLOCATED, o1p1.getContainer().getState()); Assert.assertEquals(PTOperator.State.PENDING_DEPLOY, o1p1.getState()); ContainerStats cstats = new ContainerStats(containerId); ContainerHeartbeat hb = new ContainerHeartbeat(); hb.setContainerStats(cstats); ContainerHeartbeatResponse chr = scm.processHeartbeat(hb); Assert.assertNotNull(chr.deployRequest); Assert.assertEquals("" + chr.deployRequest, 1, chr.deployRequest.size()); Assert.assertEquals(PTContainer.State.ACTIVE, o1p1.getContainer().getState()); Assert.assertEquals("state " + o1p1, PTOperator.State.PENDING_DEPLOY, o1p1.getState()); OperatorHeartbeat ohb = new OperatorHeartbeat(); ohb.setNodeId(o1p1.getId()); ohb.setState(OperatorHeartbeat.DeployState.ACTIVE); OperatorStats stats = new OperatorStats(); stats.checkpoint = new Checkpoint(2, 0, 0); stats.windowId = 3; stats.outputPorts = Lists.newArrayList(); PortStats ps = new PortStats(TestGeneratorInputOperator.OUTPUT_PORT); ps.bufferServerBytes = 101; ps.tupleCount = 1; stats.outputPorts.add(ps); ohb.windowStats = Lists.newArrayList(stats); cstats.operators.add(ohb); scm.processHeartbeat(hb); Assert.assertEquals(PTContainer.State.ACTIVE, o1p1.getContainer().getState()); Assert.assertEquals("state " + o1p1, PTOperator.State.ACTIVE, o1p1.getState()); Assert.assertEquals("tuples " + o1p1, 1, o1p1.stats.totalTuplesEmitted.get()); Assert.assertEquals("tuples " + o1p1, 0, o1p1.stats.totalTuplesProcessed.get()); Assert.assertEquals("window " + o1p1, 3, o1p1.stats.currentWindowId.get()); Assert.assertEquals("port stats", 1, o1p1.stats.outputPortStatusList.size()); PortStatus o1p1ps = o1p1.stats.outputPortStatusList.get(TestGeneratorInputOperator.OUTPUT_PORT); Assert.assertNotNull("port stats", o1p1ps); Assert.assertEquals("port stats", 1, o1p1ps.totalTuples); stats = new OperatorStats(); stats.checkpoint = new Checkpoint(2, 0, 0); stats.windowId = 4; stats.outputPorts = Lists.newArrayList(); ps = new PortStats(TestGeneratorInputOperator.OUTPUT_PORT); ps.bufferServerBytes = 1; ps.tupleCount = 1; stats.outputPorts.add(ps); ohb.windowStats = Lists.newArrayList(stats); cstats.operators.clear(); cstats.operators.add(ohb); scm.processHeartbeat(hb); Assert.assertEquals("tuples " + o1p1, 2, o1p1.stats.totalTuplesEmitted.get()); Assert.assertEquals("window " + o1p1, 4, o1p1.stats.currentWindowId.get()); Assert.assertEquals("statsQueue " + o1p1, 2, o1p1.stats.listenerStats.size()); scm.processEvents(); Assert.assertEquals("statsQueue " + o1p1, 0, o1p1.stats.listenerStats.size()); Assert.assertEquals("lastStats " + o1p1, 2, o1p1.stats.lastWindowedStats.size()); }