@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 shouldOutputChangeStreamMutations() {
    ChangeStreamMutation mutation = mock(ChangeStreamMutation.class);
    doFn.processElement(KV.of(ByteString.copyFromUtf8("test"), mutation), outputReceiver);
    verify(outputReceiver, times(1)).output(KV.of(ByteString.copyFromUtf8("test"), mutation));
  }