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;
    }
@Test
    public void testHeadersWithCustomConverter() throws Exception {
        StringConverter stringConverter = new StringConverter();
        SampleConverterWithHeaders testConverter = new SampleConverterWithHeaders();

        createWorkerTask(stringConverter, testConverter, stringConverter, RetryWithToleranceOperatorTest.noopOperator(),
                Collections::emptyList);

        expectSendRecord(null);
        expectApplyTransformationChain();
        expectTopicCreation(TOPIC);

        String stringA = "Árvíztűrő tükörfúrógép";
        String encodingA = "latin2";
        String stringB = "Тестовое сообщение";
        String encodingB = "koi8_r";

        org.apache.kafka.connect.header.Headers headersA = new ConnectHeaders()
            .addString("encoding", encodingA);
        org.apache.kafka.connect.header.Headers headersB = new ConnectHeaders()
            .addString("encoding", encodingB);

        workerTask.toSend = Arrays.asList(
            new SourceRecord(PARTITION, OFFSET, "topic", null, Schema.STRING_SCHEMA, "a",
                Schema.STRING_SCHEMA, stringA, null, headersA),
            new SourceRecord(PARTITION, OFFSET, "topic", null, Schema.STRING_SCHEMA, "b",
                Schema.STRING_SCHEMA, stringB, null, headersB)
        );
        workerTask.sendRecords();

        ArgumentCaptor<ProducerRecord<byte[], byte[]>> sent = verifySendRecord(2);

        List<ProducerRecord<byte[], byte[]>> capturedValues = sent.getAllValues();
        assertEquals(2, capturedValues.size());

        ProducerRecord<byte[], byte[]> sentRecordA = capturedValues.get(0);
        ProducerRecord<byte[], byte[]> sentRecordB = capturedValues.get(1);

        assertEquals(ByteBuffer.wrap("a".getBytes()), ByteBuffer.wrap(sentRecordA.key()));
        assertEquals(
            ByteBuffer.wrap(stringA.getBytes(encodingA)),
            ByteBuffer.wrap(sentRecordA.value())
        );
        assertEquals(encodingA, new String(sentRecordA.headers().lastHeader("encoding").value()));

        assertEquals(ByteBuffer.wrap("b".getBytes()), ByteBuffer.wrap(sentRecordB.key()));
        assertEquals(
            ByteBuffer.wrap(stringB.getBytes(encodingB)),
            ByteBuffer.wrap(sentRecordB.value())
        );
        assertEquals(encodingB, new String(sentRecordB.headers().lastHeader("encoding").value()));

        verifyTaskGetTopic(2);
        verifyTopicCreation();
    }