HiveMQ Kafka Extension Customization

public class BusinessLogicTransformer implements MqttToKafkaTransformer {

    @Override
    public void transformMqttToKafka(
            final @NotNull MqttToKafkaInput input,
            final @NotNull MqttToKafkaOutput output) {
        // Get MQTT payload
        final ByteBuffer  mqttPayload= input.getPublishPacket().getPayload().orElse(ByteBuffer.allocate(0));

        // Manipulate the MQTT payload according to your business logic
        // For example, deserialize or add fields
        final byte[] recordValue = ownBusinessLogic(mqttPayload);

        final KafkaRecord kafkaRecord = output.newKafkaRecordBuilder()
                .topic(input.getPublishPacket().getTopic())
                .key(input.getPublishPacket().getResponseTopic().orElse("unknown"))
                .value(recordValue)
                .build();

        // Set Record as output
        output.setKafkaRecords(List.of(kafkaRecord));
    }
}

public class MulticastTransformer implements MqttToKafkaTransformer {

    private static final String KAFKA_TOPIC_LOCATION_SERVICE = "location-service";
    private static final String KAFKA_TOPIC_SPEED_SERVICE = "speed-control-service";
    private static final String KAFKA_TOPIC_FUEL_SERVICE = "fuel-control-service";

    @Override
    public void transformMqttToKafka(
            final @NotNull MqttToKafkaInput input,
            final @NotNull MqttToKafkaOutput output) {
        // Get MQTT payload
        final ByteBuffer mqttPayload = input.getPublishPacket().getPayload().get();
        // Extract information from MQTT payload
        final byte[] location = extractLocation(mqttPayload);
        // Create record for the location information
        final KafkaRecord locationRecord = output.newKafkaRecordBuilder()
                .topic(KAFKA_TOPIC_LOCATION_SERVICE)
                .key(input.getPublishPacket().getTopic())
                .value(location)
                .build();
        // Create record for the speed information
        final byte[] speed = extractSpeed(mqttPayload);
        final KafkaRecord speedRecord = output.newKafkaRecordBuilder()
                .topic(KAFKA_TOPIC_SPEED_SERVICE)
                .key(input.getPublishPacket().getTopic())
                .value(speed)
                .build();
        // Create record for the fuel information
        final byte[] fuel = extractFuel(mqttPayload);
        final KafkaRecord fuelRecord = output.newKafkaRecordBuilder()
                .topic(KAFKA_TOPIC_FUEL_SERVICE)
                .key(input.getPublishPacket().getTopic())
                .value(fuel)
                .build();

        // Set the 3 Kafka records as output
        output.setKafkaRecords(List.of(locationRecord, speedRecord, fuelRecord));
    }

    private byte[] extractFuel(final @NotNull ByteBuffer mqttPayload) {
        // Pseudo-implementation, add custom code
        return null;
    }

    private byte[] extractSpeed(final @NotNull ByteBuffer mqttPayload) {
        // Pseudo-implementation, add custom code
        return null;
    }

    private byte[] extractLocation(final @NotNull ByteBuffer mqttPayload) {
        // Pseudo-implementation, add custom code
        return null;
    }
}

public class DroppingTransformer implements MqttToKafkaTransformer {

    @Override
    public void transformMqttToKafka(
            final @NotNull MqttToKafkaInput input,
            final @NotNull MqttToKafkaOutput output) {
        // If no payload is set, drop message
        if (input.getPublishPacket().getPayload().isEmpty()) {
            output.setKafkaRecords(List.of());
            return;
        }

        // Create simple Kafka record
        final KafkaRecord kafkaRecord = output.newKafkaRecordBuilder()
                .topic(input.getPublishPacket().getTopic())
                .key(input.getPublishPacket().getResponseTopic().orElse("unknown"))
                .value(input.getPublishPacket().getPayload().get())
                .build();

        // Set Record as output
        output.setKafkaRecords(List.of(kafkaRecord));
    }
}

public class CreateTopicsTransformer implements MqttToKafkaTransformer {

    private static final @NotNull Logger log = LoggerFactory.getLogger(CreateTopicsTransformer.class);

    public static final String KAFKA_TOPIC = "example-topic";

    @Override
    public void init(final @NotNull MqttToKafkaInitInput input) {
        // Get the service for Kafka topic
        final KafkaTopicService kafkaTopicService = input.getKafkaTopicService();

        // See if Kafka topics exist
        final KafkaTopicService.KafkaTopicState topicState = kafkaTopicService.getKafkaTopicState(KAFKA_TOPIC);

        if (topicState== KafkaTopicService.KafkaTopicState.MISSING) {
            final KafkaTopicService.KafkaTopicState result = kafkaTopicService.createKafkaTopic(KAFKA_TOPIC);
            if (result == KafkaTopicService.KafkaTopicState.CREATED) {
                log.info("Topic '{}' was created successfully", KAFKA_TOPIC);
            } else {
                log.warn("Topic '{}' could not be created", KAFKA_TOPIC);
            }
        } else if (topicState == KafkaTopicService.KafkaTopicState.FAILURE) {
            // The query on the topic failed (handle this error by logging or custom logic)
        }
    }

    @Override
    public void transformMqttToKafka(
            final @NotNull MqttToKafkaInput input,
            final @NotNull MqttToKafkaOutput output) {
        // Get MQTT payload
        final ByteBuffer mqttPayload= input.getPublishPacket().getPayload().orElse(ByteBuffer.allocate(0));

        final KafkaRecord kafkaRecord = output.newKafkaRecordBuilder()
                .topic(KAFKA_TOPIC)
                .key(input.getPublishPacket().getTopic())
                .value(mqttPayload)
                .build();

        // Set Record as output
        output.setKafkaRecords(List.of(kafkaRecord));
    }
}

public class CustomTransformer implements KafkaToMqttTransformer {

    @Override
    public void transformKafkaToMqtt(
            final @NotNull KafkaToMqttInput input,
            final @NotNull KafkaToMqttOutput output) {
        // Get the Kafka record
        final KafkaRecord kafkaRecord = input.getKafkaRecord();

        // Get value of Kafka record and use it to build MQTT payload according to custom business logic
        final byte[] kafkaValue = kafkaRecord.getValueAsByteArray().orElse(new byte[0]);
        final byte[] mqttPayload = ownBusinessLogic(kafkaValue);

        // Get the Kafka key and use it as the MQTT topic
        final byte[] kafkaKey = kafkaRecord.getKeyAsByteArray().orElse("unknown".getBytes());
        final String mqttTopic = new String(kafkaKey);

        // Build MQTT publish
        final Publish publish = output.newPublishBuilder()
                .topic(mqttTopic)
                .qos(Qos.AT_LEAST_ONCE)
                .payload(ByteBuffer.wrap(mqttPayload))
                .build();

        // Set MQTT publish as output
        output.setPublishes(List.of(publish));
    }

    private byte @NotNull [] ownBusinessLogic(final byte @NotNull [] mqttPayload) {
        // Insert your own business logic
        return mqttPayload;
    }
}

public class SkippingTransformer implements KafkaToMqttTransformer {

    private static final Logger log = LoggerFactory.getLogger(SkippingTransformer.class);

    @Override
    public void transformKafkaToMqtt(
            final @NotNull KafkaToMqttInput input,
            final @NotNull KafkaToMqttOutput output) {
        // Get the Kafka record
        final KafkaRecord kafkaRecord = input.getKafkaRecord();

        // Skip the Kafka record if the key or value are missing
        if (kafkaRecord.getKey().isEmpty() || kafkaRecord.getValue().isEmpty()) {
            log.warn("Value or Key were empty, skipping Kafka record");
            output.setPublishes(List.of());
            return;
        }

        // Get value of Kafka record and use it to build MQTT payload according to custom business logic
        final byte[] mqttPayload = kafkaRecord.getValueAsByteArray().orElse(new byte[0]);

        // Get the Kafka key and use it as the MQTT topic
        final byte[] kafkaKey = kafkaRecord.getKeyAsByteArray().orElse("unknown".getBytes());
        final String mqttTopic = new String(kafkaKey);

        // Build MQTT publish
        final Publish publish = output.newPublishBuilder()
                .topic(mqttTopic)
                .qos(Qos.AT_LEAST_ONCE)
                .payload(ByteBuffer.wrap(mqttPayload))
                .build();

        // Set MQTT Publish as output
        output.setPublishes(List.of(publish));
        output.setPublishes(List.of(publish));
    }
}

public class ThirdPartyDataTransformer implements KafkaToMqttTransformer {

    @Override
    public void transformKafkaToMqtt(
            final @NotNull KafkaToMqttInput input,
            final @NotNull KafkaToMqttOutput output) {
        // Get the Kafka record to get value und key
        final KafkaRecord kafkaRecord = input.getKafkaRecord();
        final byte[] kafkaKey = kafkaRecord.getKeyAsByteArray().get();
        final byte[] kafkaValue = kafkaRecord.getValueAsByteArray().get();

        // Use the key and value of the Kafka Record to load information from another source and use it as MQTT Payload
        final byte[] mqttPayload = enrichKafkaValue(kafkaKey, kafkaValue);

        // Build MQTT publish
        final Publish publish = output.newPublishBuilder()
                .topic(kafkaRecord.getTopic())
                .qos(Qos.AT_LEAST_ONCE)
                .payload(ByteBuffer.wrap(mqttPayload))
                .build();

        // Set MQTT Publish as output
        output.setPublishes(List.of(publish));
    }

    private byte @NotNull [] enrichKafkaValue(
            final byte @NotNull [] kafkaKey,
            final byte @NotNull [] kafkaValue) {
        // Open files/db or contact other systems here and build MQTT payload or other information
        final byte[] mqttPayload = new byte[0];
        return mqttPayload;
    }
}