@Test
  public void testTaskApplication() throws Exception {
    JobGraphJsonGenerator jsonGenerator = new JobGraphJsonGenerator();
    JobGraph mockJobGraph = mock(JobGraph.class);
    ApplicationConfig mockAppConfig = mock(ApplicationConfig.class);
    when(mockAppConfig.getAppName()).thenReturn("testTaskApp");
    when(mockAppConfig.getAppId()).thenReturn("testTaskAppId");
    when(mockJobGraph.getApplicationConfig()).thenReturn(mockAppConfig);
    // compute the three disjoint sets of the JobGraph: input only, output only, and intermediate streams
    Set<StreamEdge> inEdges = new HashSet<>(mockJobNode.getInEdges().values());
    Set<StreamEdge> outEdges = new HashSet<>(mockJobNode.getOutEdges().values());
    Set<StreamEdge> intermediateEdges = new HashSet<>(inEdges);
    // intermediate streams are the intersection between input and output
    intermediateEdges.retainAll(outEdges);
    // remove all intermediate streams from input
    inEdges.removeAll(intermediateEdges);
    // remove all intermediate streams from output
    outEdges.removeAll(intermediateEdges);
    // set the return values for mockJobGraph
    when(mockJobGraph.getInputStreams()).thenReturn(inEdges);
    when(mockJobGraph.getOutputStreams()).thenReturn(outEdges);
    when(mockJobGraph.getIntermediateStreamEdges()).thenReturn(intermediateEdges);
    Set<TableDescriptor> tables = new HashSet<>(mockJobNode.getTables().values());
    when(mockJobGraph.getTables()).thenReturn(tables);
    when(mockJobGraph.getJobNodes()).thenReturn(Collections.singletonList(mockJobNode));
    String graphJson = jsonGenerator.toJson(mockJobGraph);
    ObjectMapper objectMapper = new ObjectMapper();
    JobGraphJsonGenerator.JobGraphJson jsonObject = objectMapper.readValue(graphJson.getBytes(), JobGraphJsonGenerator.JobGraphJson.class);
    assertEquals("testTaskAppId", jsonObject.applicationId);
    assertEquals("testTaskApp", jsonObject.applicationName);
    Set<String> inStreamIds = inEdges.stream().map(stream -> stream.getStreamSpec().getId()).collect(Collectors.toSet());
    assertThat(jsonObject.sourceStreams.keySet(), Matchers.containsInAnyOrder(inStreamIds.toArray()));
    Set<String> outStreamIds = outEdges.stream().map(stream -> stream.getStreamSpec().getId()).collect(Collectors.toSet());
    assertThat(jsonObject.sinkStreams.keySet(), Matchers.containsInAnyOrder(outStreamIds.toArray()));
    Set<String> intStreamIds = intermediateEdges.stream().map(stream -> stream.getStreamSpec().getId()).collect(Collectors.toSet());
    assertThat(jsonObject.intermediateStreams.keySet(), Matchers.containsInAnyOrder(intStreamIds.toArray()));
    Set<String> tableIds = tables.stream().map(t -> t.getTableId()).collect(Collectors.toSet());
    assertThat(jsonObject.tables.keySet(), Matchers.containsInAnyOrder(tableIds.toArray()));
    JobGraphJsonGenerator.JobNodeJson expectedNodeJson = new JobGraphJsonGenerator.JobNodeJson();
    expectedNodeJson.jobId = mockJobNode.getJobId();
    expectedNodeJson.jobName = mockJobNode.getJobName();
    assertEquals(1, jsonObject.jobs.size());
    JobGraphJsonGenerator.JobNodeJson actualNodeJson = jsonObject.jobs.get(0);
    assertEquals(expectedNodeJson.jobId, actualNodeJson.jobId);
    assertEquals(expectedNodeJson.jobName, actualNodeJson.jobName);
    assertEquals(3, actualNodeJson.operatorGraph.inputStreams.size());
    assertEquals(2, actualNodeJson.operatorGraph.outputStreams.size());
    assertEquals(0, actualNodeJson.operatorGraph.operators.size());
  }
the graph looks like the following.
number in parentheses () indicates number of stream partitions.
number in parentheses in quotes ("") indicates expected partition count.
number in square brackets [] indicates operator ID.

input3 (32) -> filter [7] -> partitionBy [8] ("64") -> map [10] -> join [14] -> sendTo(output2) [15] (16)
                                                                  |
             input2 (16) -> partitionBy [3] ("64") -> filter [5] -| -> sink [13]
                                                                  |
                                        input1 (64) -> map [1] -> join [11] -> sendTo(output1) [12] (8)
samza-core/src/test/java/org/apache/samza/execution/TestJobGraphJsonGenerator.java