@Override
    protected Optional<SubmittedRecords.SubmittedRecord> prepareToSendRecord(
            SourceRecord sourceRecord,
            ProducerRecord<byte[], byte[]> producerRecord
    ) {
        if (offsetStore.primaryOffsetsTopic().equals(producerRecord.topic())) {
            // This is to prevent deadlock that occurs when:
            //     1. A task provides a record whose topic is the task's offsets topic
            //     2. That record is dispatched to the task's producer in a transaction that remains open
            //        at least until the worker polls the task again
            //     3. In the subsequent call to SourceTask::poll, the task requests offsets from the worker
            //        (which requires a read to the end of the offsets topic, and will block until any open
            //        transactions on the topic are either committed or aborted)
            throw new ConnectException("Source tasks may not produce to their own offsets topics when exactly-once support is enabled");
        }
        maybeBeginTransaction();
        return Optional.empty();
    }
WorkerTask that uses a {@link SourceTask} to ingest data into Kafka, with support for exactly-once semantics.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java