@ProcessElement
  public void processElement(
      @Element KV<ByteString, ChangeStreamRecord> changeStreamRecordKV,
      OutputReceiver<KV<ByteString, ChangeStreamMutation>> receiver) {
    ChangeStreamRecord inputRecord = changeStreamRecordKV.getValue();
    if (inputRecord instanceof ChangeStreamMutation) {
      receiver.output(KV.of(changeStreamRecordKV.getKey(), (ChangeStreamMutation) inputRecord));
    }
  }
@Test
  public void shouldOutputCloseStreams() {
    // This shouldn't happen but if it were to we wouldn't want the CloseStreams to be returned to
    // users
    CloseStream closeStream = mock(CloseStream.class);
    doFn.processElement(KV.of(ByteString.copyFromUtf8("test"), closeStream), outputReceiver);
    verify(outputReceiver, never()).output(any());
  }