@ProcessElement
  public void processElement(
      @Element DataChangeRecord dataChangeRecord, OutputReceiver<DataChangeRecord> receiver) {
    final Instant commitInstant =
        new Instant(dataChangeRecord.getCommitTimestamp().toSqlTimestamp().getTime());

    metrics.incDataRecordCounter();
    measureCommitTimestampToEmittedMillis(dataChangeRecord);
    measureStreamMillis(dataChangeRecord);

    receiver.outputWithTimestamp(dataChangeRecord, commitInstant);
  }
@Test
  public void testPostProcessingMetrics() {
    DataChangeRecord dataChangeRecord =
        new DataChangeRecord(
            "partitionToken",
            Timestamp.ofTimeMicroseconds(1L),
            "serverTransactionId",
            true,
            "recordSequence",
            "tableName",
            Arrays.asList(
                new ColumnType("column1", new TypeCode("type1"), true, 1L),
                new ColumnType("column2", new TypeCode("type2"), false, 2L)),
            Collections.singletonList(
                new Mod(
                    "{\"column1\": \"value1\"}",
                    "{\"column2\": \"oldValue2\"}",
                    "{\"column2\": \"newValue2\"}")),
            ModType.UPDATE,
            ValueCaptureType.OLD_AND_NEW_VALUES,
            10L,
            2L,
            "transactionTag",
            true,
            ChangeStreamRecordMetadata.newBuilder()
                .withRecordStreamStartedAt(Timestamp.ofTimeMicroseconds(1L))
                .withRecordStreamEndedAt(Timestamp.ofTimeMicroseconds(2L))
                .build());
    doNothing().when(changeStreamMetrics).incDataRecordCounter();
    doNothing().when(changeStreamMetrics).updateDataRecordCommittedToEmitted(any());
    processingMetricsDoFn.processElement(dataChangeRecord, receiver);
    verify(changeStreamMetrics, times(1)).incDataRecordCounter();
    verify(changeStreamMetrics, times(1)).updateDataRecordCommittedToEmitted(any());
  }