@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 shouldRetainOriginalTopicPartition() {
        String transformedTopic = "transformed-test-topic";
        SinkRecord sinkRecord = new SinkRecord(transformedTopic, 0, null, null, null, null, 10);
        ConsumerRecord<byte[], byte[]> consumerRecord = new ConsumerRecord<>(TOPIC, 0, 10, null, null);
        ProcessingContext<ConsumerRecord<byte[], byte[]>> context = new ProcessingContext<>(consumerRecord);
        InternalSinkRecord internalSinkRecord = new InternalSinkRecord(context, sinkRecord);

        assertEquals(TOPIC, internalSinkRecord.originalTopic());
        assertEquals(0, internalSinkRecord.originalKafkaPartition().intValue());

        SinkRecord transformedSinkRecord = internalSinkRecord.newRecord(transformedTopic, 1, null, null, null, null, null);
        assertEquals(TOPIC, transformedSinkRecord.originalTopic());
        assertEquals(0, transformedSinkRecord.originalKafkaPartition().intValue());
    }