@Override
    public SinkRecord newRecord(String topic, Integer kafkaPartition, Schema keySchema, Object key,
                                Schema valueSchema, Object value, Long timestamp,
                                Iterable<Header> headers) {
        return new InternalSinkRecord(context, topic, kafkaPartition, keySchema, key,
                valueSchema, value, kafkaOffset(), timestamp, timestampType(), headers);
    }
@Test
    public void testNewRecordHeaders() {
        SinkRecord sinkRecord = new SinkRecord(TOPIC, 0, null, null, null, null, 10);
        ConsumerRecord<byte[], byte[]> consumerRecord = new ConsumerRecord<>("test-topic", 0, 10, null, null);
        ProcessingContext<ConsumerRecord<byte[], byte[]>> context = new ProcessingContext<>(consumerRecord);
        InternalSinkRecord internalSinkRecord = new InternalSinkRecord(context, sinkRecord);
        assertTrue(internalSinkRecord.headers().isEmpty());
        assertTrue(sinkRecord.headers().isEmpty());

        SinkRecord newRecord = internalSinkRecord.newRecord(TOPIC, 0, null, null, null,
                null, null, Collections.singletonList(mock(Header.class)));
        assertEquals(1, newRecord.headers().size());
    }