boolean sendRecords() {
        int processed = 0;
        recordBatch(toSend.size());
        final SourceRecordWriteCounter counter =
                toSend.isEmpty() ? null : new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup);
        for (final SourceRecord preTransformRecord : toSend) {
            ProcessingContext<SourceRecord> context = new ProcessingContext<>(preTransformRecord);
            final SourceRecord record = transformationChain.apply(context, preTransformRecord);
            final ProducerRecord<byte[], byte[]> producerRecord = convertTransformedRecord(context, record);
            if (producerRecord == null || context.failed()) {
                counter.skipRecord();
                recordDropped(preTransformRecord);
                processed++;
                continue;
            }

            log.trace("{} Appending record to the topic {} with key {}, value {}", this, record.topic(), record.key(), record.value());
            Optional<SubmittedRecords.SubmittedRecord> submittedRecord = prepareToSendRecord(preTransformRecord, producerRecord);
            try {
                final String topic = producerRecord.topic();
                maybeCreateTopic(topic);
                producer.send(
                    producerRecord,
                    (recordMetadata, e) -> {
                        if (e != null) {
                            if (producerClosed) {
                                log.trace("{} failed to send record to {}; this is expected as the producer has already been closed", AbstractWorkerSourceTask.this, topic, e);
                            } else {
                                log.error("{} failed to send record to {}: ", AbstractWorkerSourceTask.this, topic, e);
                            }
                            log.trace("{} Failed record: {}", AbstractWorkerSourceTask.this, preTransformRecord);
                            producerSendFailed(context, false, producerRecord, preTransformRecord, e);
                            if (retryWithToleranceOperator.getErrorToleranceType() == ToleranceType.ALL) {
                                counter.skipRecord();
                                submittedRecord.ifPresent(SubmittedRecords.SubmittedRecord::ack);
                            }
                        } else {
                            counter.completeRecord();
                            log.trace("{} Wrote record successfully: topic {} partition {} offset {}",
                                    AbstractWorkerSourceTask.this,
                                    recordMetadata.topic(), recordMetadata.partition(),
                                    recordMetadata.offset());
                            recordSent(preTransformRecord, producerRecord, recordMetadata);
                            submittedRecord.ifPresent(SubmittedRecords.SubmittedRecord::ack);
                            if (topicTrackingEnabled) {
                                recordActiveTopic(producerRecord.topic());
                            }
                        }
                    });
                // Note that this will cause retries to take place within a transaction
            } catch (RetriableException | org.apache.kafka.common.errors.RetriableException e) {
                log.warn("{} Failed to send record to topic '{}' and partition '{}'. Backing off before retrying: ",
                        this, producerRecord.topic(), producerRecord.partition(), e);
                toSend = toSend.subList(processed, toSend.size());
                submittedRecord.ifPresent(SubmittedRecords.SubmittedRecord::drop);
                counter.retryRemaining();
                return false;
            } catch (ConnectException e) {
                log.warn("{} Failed to send record to topic '{}' and partition '{}' due to an unrecoverable exception: ",
                        this, producerRecord.topic(), producerRecord.partition(), e);
                log.trace("{} Failed to send {} with unrecoverable exception: ", this, producerRecord, e);
                throw e;
            } catch (KafkaException e) {
                producerSendFailed(context, true, producerRecord, preTransformRecord, e);
            }
            processed++;
            recordDispatched(preTransformRecord);
        }
        toSend = null;
        batchDispatched();
        return true;
    }
Try to send a batch of records. If a send fails and is retriable, this saves the remainder of the batch so it can
be retried after backing off. If a send fails and is not retriable, this will throw a ConnectException.
@return true if all messages were sent, false if some need to be retried
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java