@Override
    protected Optional<SubmittedRecords.SubmittedRecord> prepareToSendRecord(
            SourceRecord sourceRecord,
            ProducerRecord<byte[], byte[]> producerRecord
    ) {
        maybeThrowProducerSendException();
        return Optional.of(submittedRecords.submit(sourceRecord));
    }
{@link WorkerTask} that uses a {@link SourceTask} to ingest data into Kafka.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java