@VisibleForTesting
  static Map<String, Map.Entry<String, String>> buildTransformIOMap(
      Pipeline pipeline, ConfigContext ctx) {
    final Map<String, Map.Entry<String, String>> pTransformToInputOutputMap = new HashMap<>();
    final SamzaPipelineTranslator.TransformVisitorFn configFn =
        new SamzaPipelineTranslator.TransformVisitorFn() {
          @Override
          public <T extends PTransform<?, ?>> void apply(
              T transform,
              TransformHierarchy.Node node,
              Pipeline pipeline,
              TransformTranslator<T> translator) {
            ctx.setCurrentTransform(node.toAppliedPTransform(pipeline));
            List<String> inputs = getIOPValueList(node.getInputs()).get();
            List<String> outputs = getIOPValueList(node.getOutputs()).get();
            pTransformToInputOutputMap.put(
                node.getFullName(),
                new AbstractMap.SimpleEntry<>(
                    String.join(TRANSFORM_IO_MAP_DELIMITER, inputs),
                    String.join(TRANSFORM_IO_MAP_DELIMITER, outputs)));
            ctx.clearCurrentTransform();
          }
        };

    final SamzaPipelineTranslator.SamzaPipelineVisitor visitor =
        new SamzaPipelineTranslator.SamzaPipelineVisitor(configFn);
    pipeline.traverseTopologically(visitor);
    return pTransformToInputOutputMap;
  }
@Test
  public void testBeamTransformIOConfigGen() {
    SamzaPipelineOptions options = PipelineOptionsFactory.create().as(SamzaPipelineOptions.class);
    options.setJobName("TestEnvConfig");
    options.setRunner(SamzaRunner.class);
    options.setSamzaExecutionEnvironment(SamzaExecutionEnvironment.LOCAL);

    Pipeline pipeline = Pipeline.create(options);
    pipeline.apply(Impulse.create()).apply(Filter.by(Objects::nonNull));
    pipeline.replaceAll(SamzaTransformOverrides.getDefaultOverrides());

    final Map<PValue, String> idMap = PViewToIdMapper.buildIdMap(pipeline);
    final Set<String> nonUniqueStateIds = StateIdParser.scan(pipeline);
    final ConfigContext configCtx = new ConfigContext(idMap, nonUniqueStateIds, options);

    final ConfigBuilder configBuilder = new ConfigBuilder(options);
    SamzaPipelineTranslator.createConfig(pipeline, configCtx, configBuilder);
    final Map<String, Map.Entry<String, String>> transformInputOutput =
        PipelineJsonRenderer.buildTransformIOMap(pipeline, configCtx);

    assertEquals(2, transformInputOutput.size());
    assertEquals("", transformInputOutput.get("Impulse").getKey()); // no input to impulse
    assertEquals(
        "Impulse.out",
        transformInputOutput.get("Impulse").getValue()); // PValue for to Impulse.output

    // Input to Filter is PValue Output from Impulse
    assertEquals(
        "Impulse.out",
        transformInputOutput.get("Filter/ParDo(Anonymous)/ParMultiDo(Anonymous)").getKey());
    // output PValue of filter
    assertEquals(
        "Filter/ParDo(Anonymous)/ParMultiDo(Anonymous).output",
        transformInputOutput.get("Filter/ParDo(Anonymous)/ParMultiDo(Anonymous)").getValue());
  }