@Override
    public R apply(R record) {
        final Matcher matcher = regex.matcher(record.topic());
        if (matcher.matches()) {
            final String topic = matcher.replaceFirst(replacement);
            log.trace("Rerouting from topic '{}' to new topic '{}'", record.topic(), topic);
            return record.newRecord(topic, record.kafkaPartition(), record.keySchema(), record.key(), record.valueSchema(), record.value(), record.timestamp());
        } else {
            log.trace("Not rerouting topic '{}' as it does not match the configured regex", record.topic());
        }
        return record;
    }
@Test
    public void addPrefix() {
        assertEquals("prefix-orig", apply("(.*)", "prefix-$1", "orig"));
    }