title: Global combine not producing output Apache Beam
body: <p>I am trying to write an unbounded ping pipeline that takes output from a ping command and parses it to determine some statistics about the RTT (avg/min/max) and for now, just print the results.</p>
  <p>I have already written an unbounded ping source that outputs each line as it comes in. The results are windowed every second for every 5 seconds of pings. The windowed data is fed to a <code>Combine.globally</code> call to statefully process the string outputs. The problem is that the accumulators are never merged and the output is never extracted. This means that the pipeline never continues past this point. What am I doing wrong here?</p>
  <pre><code>public class TestPingIPs {
     public static void main(String[] args)
     {
        PipelineOptions options = PipelineOptionsFactory.create();
        Pipeline pipeline = Pipeline.create(options);
        String destination = &quot;8.8.8.8&quot;;
        PCollection&lt;PingResult&gt; res =
                /*
                Run the unbounded ping command. Only the lines where the result of the ping command are returned.
                No statistics or first startup lines are returned here.
                 */
                pipeline.apply(&quot;Ping command&quot;,
                        PingCmd.read()
                                .withPingArguments(PingCmd.PingArguments.create(destination, -1)))
               /*
               Window the ping command strings into 5 second sliding windows produced every 1 second
                */
                .apply(&quot;Window strings&quot;,
                        Window.into(SlidingWindows.of(Duration.standardSeconds(5))
                                .every(Duration.standardSeconds(1))))
               /*
               Parse and aggregate the strings into a PingResult object using stateful processing.
                */
                .apply(&quot;Combine the pings&quot;,
                        Combine.globally(new ProcessPings()).withoutDefaults())
               /*
               Test our output to see what we get here
                */
                .apply(&quot;Test output&quot;,
                        ParDo.of(new DoFn&lt;PingResult, PingResult&gt;() {
                   @ProcessElement
                   public void processElement(ProcessContext c)
                   {
                      System.out.println(c.element().getAvgRTT());
                      System.out.println(c.element().getPacketLoss());
                      c.output(c.element());
                   }
                }));
  
        pipeline.run().waitUntilFinish();
     }
  
  
     static class ProcessPings extends Combine.CombineFn&lt;String, RttStats, PingResult&gt; {
        private long getRTTFromLine(String line){
           long rtt = Long.parseLong(line.split(&quot;time=&quot;)[1].split(&quot;ms&quot;)[0]);
           return rtt;
        }
  
        @Override
        public RttStats createAccumulator()
        {
           return new RttStats();
        }
  
        @Override
        public RttStats addInput(RttStats mutableAccumulator, String input)
        {
           mutableAccumulator.incTotal();
           if (input.contains(&quot;unreachable&quot;)) {
              _unreachableCount.inc();
              mutableAccumulator.incPacketLoss();
           }
           else if (input.contains(&quot;General failure&quot;)) {
              _transmitFailureCount.inc();
              mutableAccumulator.incPacketLoss();
           }
           else if (input.contains(&quot;timed out&quot;)) {
              _timeoutCount.inc();
              mutableAccumulator.incPacketLoss();
           }
           else if (input.contains(&quot;could not find&quot;)) {
              _unknownHostCount.inc();
              mutableAccumulator.incPacketLoss();
           }
           else {
              _successfulCount.inc();
              mutableAccumulator.add(getRTTFromLine(input));
           }
  
           return mutableAccumulator;
        }
  
        @Override
        public RttStats mergeAccumulators(Iterable&lt;RttStats&gt; accumulators)
        {
           Iterator&lt;RttStats&gt; iter = accumulators.iterator();
           if (!iter.hasNext()){
              return createAccumulator();
           }
           RttStats running = iter.next();
           while (iter.hasNext()){
              RttStats next = iter.next();
              running.addAll(next.getVals());
              running.addLostPackets(next.getLostPackets());
           }
           return running;
        }
  
        @Override
        public PingResult extractOutput(RttStats stats)
        {
           stats.calculate();
           boolean connected = stats.getPacketLoss() != 1;
           return new PingResult(connected, stats.getAvg(), stats.getMin(), stats.getMax(), stats.getPacketLoss());
        }
  
        private final Counter _successfulCount = Metrics.counter(ProcessPings.class, &quot;Successful pings&quot;);
        private final Counter _unknownHostCount = Metrics.counter(ProcessPings.class, &quot;Unknown hosts&quot;);
        private final Counter _transmitFailureCount = Metrics.counter(ProcessPings.class, &quot;Transmit failures&quot;);
        private final Counter _timeoutCount = Metrics.counter(ProcessPings.class, &quot;Timeouts&quot;);
        private final Counter _unreachableCount = Metrics.counter(ProcessPings.class, &quot;Unreachable host&quot;);
     }
  </code></pre>
  <p>I would guess that there are some issues with the <code>CombineFn</code> that I wrote, but I can't seem to figure out what's going wrong here! I tried following the <a href="https://beam.apache.org/blog/stateful-processing/" rel="nofollow noreferrer">example here</a>, but there's still something I must be missing.</p>
  <p>EDIT: I added the ping command implementation below. This is running on a Direct Runner while I test.</p>
  <p>PingCmd.java:</p>
  <pre><code>public class PingCmd {
   public static Read read(){
        if (System.getProperty(&quot;os.name&quot;).startsWith(&quot;Windows&quot;)) {
           return WindowsPingCmd.read();
        }
        else{
           return null;
        }
     }
  </code></pre>
  <p>WindowsPingCmd.java:</p>
  <pre><code>public class WindowsPingCmd extends PingCmd {
     private WindowsPingCmd()
     {
     }
  
     public static PingCmd.Read read()
     {
        return new WindowsRead.Builder().build();
     }
  
  
     static class PingCheckpointMark implements UnboundedSource.CheckpointMark, Serializable {
        @VisibleForTesting
        Instant oldestMessageTimestamp = Instant.now();
        @VisibleForTesting
        transient List&lt;String&gt; outputs = new ArrayList&lt;&gt;();
  
        public PingCheckpointMark()
        {
        }
  
        public void add(String message, Instant timestamp)
        {
           if (timestamp.isBefore(oldestMessageTimestamp)) {
              oldestMessageTimestamp = timestamp;
           }
           outputs.add(message);
        }
  
        @Override
        public void finalizeCheckpoint()
        {
           oldestMessageTimestamp = Instant.now();
           outputs.clear();
        }
  
        // set an empty list to messages when deserialize
        private void readObject(java.io.ObjectInputStream stream)
                throws IOException, ClassNotFoundException
        {
           stream.defaultReadObject();
           outputs = new ArrayList&lt;&gt;();
        }
  
        @Override
        public boolean equals(@Nullable Object other)
        {
           if (other instanceof PingCheckpointMark) {
              PingCheckpointMark that = (PingCheckpointMark) other;
              return Objects.equals(this.oldestMessageTimestamp, that.oldestMessageTimestamp)
                      &amp;&amp; Objects.deepEquals(this.outputs, that.outputs);
           }
           else {
              return false;
           }
        }
     }
  
  
     @VisibleForTesting
     static class UnboundedPingSource extends UnboundedSource&lt;String, PingCheckpointMark&gt; {
  
        private final WindowsRead spec;
  
        public UnboundedPingSource(WindowsRead spec)
        {
           this.spec = spec;
        }
  
        @Override
        public UnboundedReader&lt;String&gt; createReader(
                PipelineOptions options, PingCheckpointMark checkpointMark)
        {
           return new UnboundedPingReader(this, checkpointMark);
        }
  
        @Override
        public List&lt;UnboundedPingSource&gt; split(int desiredNumSplits, PipelineOptions options)
        {
           // Don't really need to ever split the ping source, so we should just have one per destination
           return Collections.singletonList(new UnboundedPingSource(spec));
        }
  
        @Override
        public void populateDisplayData(DisplayData.Builder builder)
        {
           spec.populateDisplayData(builder);
        }
  
        @Override
        public Coder&lt;PingCheckpointMark&gt; getCheckpointMarkCoder()
        {
           return SerializableCoder.of(PingCheckpointMark.class);
        }
  
        @Override
        public Coder&lt;String&gt; getOutputCoder()
        {
           return StringUtf8Coder.of();
        }
     }
  
  
     @VisibleForTesting
        static class UnboundedPingReader extends UnboundedSource.UnboundedReader&lt;String&gt; {
  
        private final UnboundedPingSource source;
  
        private String current;
        private Instant currentTimestamp;
        private final PingCheckpointMark checkpointMark;
        private BufferedReader processOutput;
        private Process process;
        private boolean finishedPings;
        private int maxCount = 5;
        private static AtomicInteger currCount = new AtomicInteger(0);
  
        public UnboundedPingReader(UnboundedPingSource source, PingCheckpointMark checkpointMark)
        {
           this.finishedPings = false;
           this.source = source;
           this.current = null;
           if (checkpointMark != null) {
              this.checkpointMark = checkpointMark;
           }
           else {
              this.checkpointMark = new PingCheckpointMark();
           }
        }
  
        @Override
        public boolean start() throws IOException
        {
           WindowsRead spec = source.spec;
           String cmd = createCommand(spec.pingConfiguration().getPingCount(), spec.pingConfiguration().getDestination());
           try {
              ProcessBuilder builder = new ProcessBuilder(cmd.split(&quot; &quot;));
              builder.redirectErrorStream(true);
              process = builder.start();
  
              processOutput = new BufferedReader(new InputStreamReader(process.getInputStream()));
              return advance();
           } catch (Exception e) {
              throw new IOException(e);
           }
        }
  
        private String createCommand(int count, String dest){
           StringBuilder builder = new StringBuilder(&quot;ping&quot;);
           String countParam = &quot;&quot;;
           if (count &lt;= 0){
              countParam = &quot;-t&quot;;
           }
           else{
              countParam += &quot;-n &quot; + count;
           }
  
           return builder.append(&quot; &quot;).append(countParam).append(&quot; &quot;).append(dest).toString();
        }
  
        @Override
        public boolean advance() throws IOException
        {
           String line = processOutput.readLine();
           // Ignore empty/null lines
           if (line == null || line.isEmpty()) {
              line = processOutput.readLine();
           }
           // Ignore the 'Pinging &lt;dest&gt; with 32 bytes of data' line
           if (line.contains(&quot;Pinging &quot; + source.spec.pingConfiguration().getDestination())) {
              line = processOutput.readLine();
           }
           // If the pings have finished, ignore
           if (finishedPings) {
              return false;
           }
           // If this is the start of the statistics, the pings are done and we can just exit
           if (line.contains(&quot;statistics&quot;)) {
              finishedPings = true;
           }
  
           current = line;
           currentTimestamp = Instant.now();
           checkpointMark.add(current, currentTimestamp);
           if (currCount.incrementAndGet() == maxCount){
              currCount.set(0);
              return false;
           }
           return true;
        }
  
        @Override
        public void close() throws IOException
        {
           if (process != null) {
              process.destroy();
              if (process.isAlive()) {
                 process.destroyForcibly();
              }
           }
        }
  
        @Override
        public Instant getWatermark()
        {
           return checkpointMark.oldestMessageTimestamp;
        }
  
        @Override
        public UnboundedSource.CheckpointMark getCheckpointMark()
        {
           return checkpointMark;
        }
  
        @Override
        public String getCurrent()
        {
           if (current == null) {
              throw new NoSuchElementException();
           }
           return current;
        }
  
        @Override
        public Instant getCurrentTimestamp()
        {
           if (current == null) {
              throw new NoSuchElementException();
           }
           return currentTimestamp;
        }
  
        @Override
        public UnboundedPingSource getCurrentSource()
        {
           return source;
        }
     }
  
  
     public static class WindowsRead extends PingCmd.Read {
        private final PingArguments pingConfig;
  
        private WindowsRead(PingArguments pingConfig)
        {
           this.pingConfig = pingConfig;
        }
  
        public Builder builder()
        {
           return new WindowsRead.Builder(this);
        }
  
        PingArguments pingConfiguration()
        {
           return pingConfig;
        }
  
        public WindowsRead withPingArguments(PingArguments configuration)
        {
           checkArgument(configuration != null, &quot;configuration can not be null&quot;);
           return builder().setPingArguments(configuration).build();
        }
  
        @Override
        public PCollection&lt;String&gt; expand(PBegin input)
        {
           org.apache.beam.sdk.io.Read.Unbounded&lt;String&gt; unbounded =
                   org.apache.beam.sdk.io.Read.from(new UnboundedPingSource(this));
  
           return input.getPipeline().apply(unbounded);
        }
  
        @Override
        public void populateDisplayData(DisplayData.Builder builder)
        {
           super.populateDisplayData(builder);
           pingConfiguration().populateDisplayData(builder);
        }
  
        static class Builder {
           private PingArguments config;
  
           Builder()
           {
           }
  
           private Builder(WindowsRead source)
           {
              this.config = source.pingConfiguration();
           }
  
           WindowsRead.Builder setPingArguments(PingArguments config)
           {
              this.config = config;
              return this;
           }
  
           WindowsRead build()
           {
              return new WindowsRead(this.config);
           }
        }
  
        @Override
        public int hashCode()
        {
           return Objects.hash(pingConfig);
        }
  
  
     }
  </code></pre>