@Override
    protected void producerSendFailed(
            ProcessingContext<SourceRecord> context,
            boolean synchronous,
            ProducerRecord<byte[], byte[]> producerRecord,
            SourceRecord preTransformRecord,
            Exception e
    ) {
        if (synchronous) {
            throw new ConnectException("Unrecoverable exception trying to send", e);
        }

        String topic = producerRecord.topic();
        if (retryWithToleranceOperator.getErrorToleranceType() == ToleranceType.ALL) {
            log.trace(
                    "Ignoring failed record send: {} failed to send record to {}: ",
                    WorkerSourceTask.this,
                    topic,
                    e
            );
            // executeFailed here allows the use of existing logging infrastructure/configuration
            retryWithToleranceOperator.executeFailed(
                    context,
                    Stage.KAFKA_PRODUCE,
                    WorkerSourceTask.class,
                    e
            );
            commitTaskRecord(preTransformRecord, null);
        } else {
            producerSendException.compareAndSet(null, e);
        }
    }
{@link WorkerTask} that uses a {@link SourceTask} to ingest data into Kafka.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java