@Test
  public void serdeTest() {
    DiagnosticsStreamMessage diagnosticsStreamMessage =
        getDiagnosticsStreamMessage(Optional.of(SAMZA_EPOCH_ID));
    Collection<DiagnosticsExceptionEvent> exceptionEventList = getExceptionList();
    diagnosticsStreamMessage.addDiagnosticsExceptionEvents(exceptionEventList);
    diagnosticsStreamMessage.addProcessorStopEvents(getProcessorStopEventList());
    diagnosticsStreamMessage.addContainerModels(getSampleContainerModels());

    MetricsSnapshot metricsSnapshot = diagnosticsStreamMessage.convertToMetricsSnapshot();
    MetricsHeader expectedHeader = new MetricsHeader(JOB_NAME, JOB_ID, CONTAINER_NAME, EXECUTION_ENV_CONTAINER_ID,
        Optional.of(SAMZA_EPOCH_ID), DiagnosticsManager.class.getName(), TASK_CLASS_VERSION, SAMZA_VERSION,
        HOSTNAME, timestamp, resetTimestamp);
    Assert.assertEquals(metricsSnapshot.getHeader(), expectedHeader);

    Map<String, Map<String, Object>> metricsMap = metricsSnapshot.getMetrics().getAsMap();
    Assert.assertTrue(metricsMap.get("org.apache.samza.container.SamzaContainerMetrics").containsKey("exceptions"));
    Assert.assertTrue(metricsMap.get(DiagnosticsManager.class.getName()).containsKey("containerModels"));
    Assert.assertTrue(metricsMap.get(DiagnosticsManager.class.getName()).containsKey("numPersistentStores"));
    Assert.assertTrue(metricsMap.get(DiagnosticsManager.class.getName()).containsKey("containerNumCores"));
    Assert.assertTrue(metricsMap.get(DiagnosticsManager.class.getName()).containsKey("containerMemoryMb"));
    Assert.assertTrue(metricsMap.get(DiagnosticsManager.class.getName()).containsKey("stopEvents"));
    Assert.assertTrue(metricsMap.get(DiagnosticsManager.class.getName()).containsKey("config"));

    DiagnosticsStreamMessage convertedDiagnosticsStreamMessage =
        DiagnosticsStreamMessage.convertToDiagnosticsStreamMessage(metricsSnapshot);
    Assert.assertEquals(convertedDiagnosticsStreamMessage, diagnosticsStreamMessage);
  }
Tests serialization and deserialization of a {@link DiagnosticsStreamMessage}
samza-core/src/test/java/org/apache/samza/diagnostics/TestDiagnosticsStreamMessage.java