private void expectConvertHeadersAndKeyValue() {
        Headers headers = new RecordHeaders();
        for (Header header : headers) {
            when(headerConverter.fromConnectHeader(eq(TOPIC), eq(header.key()), eq(Schema.STRING_SCHEMA), eq(new String(header.value()))))
                    .thenReturn(header.value());
        }
        when(keyConverter.fromConnectData(eq(TOPIC), eq(headers), eq(KEY_SCHEMA), eq(KEY)))
                .thenReturn(SERIALIZED_KEY);
        when(valueConverter.fromConnectData(eq(TOPIC), eq(headers), eq(RECORD_SCHEMA), eq(VALUE_1)))
                .thenReturn(SERIALIZED_RECORD);
    }
@return how many times an {@link OffsetStorageWriter#doFlush(Callback) offset flush} took place
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTaskTest.java