Feature/id 517/kafka to mqtt transformer (#13)

Added the KafkaToMqttTransformer interfaces.
Added get method for metric registry
Add link mechanism and extend readme

Co-authored-by: Silvio Giebl <silvio.giebl@hivemq.com>
Co-authored-by: Georg <georg.held@hivemq.com>
Co-authored-by: Silvio Giebl <silvio.giebl@hivemq.com>
Co-authored-by: Georg Held <10121608+sauroter@users.noreply.github.com>
Co-authored-by: Michael Walter <michael.walter@dc-square.de>

Author: 4 people

README.adoc

@@ -31,3 +31,11 @@ To use the SDK, add the following dependency to your Maven project:
 
 The HiveMQ Kafka Extension SDK is licensed under the `APACHE LICENSE, VERSION 2.0`.
 A copy of the license can be found link:LICENSE[here].
+
+== Setup
+To be able to use the latest changes of the `hivemq-extension-sdk` please check out the https://github.com/hivemq/hivemq-extension-sdk repository in the same folder as the current project.
+(See line `includeBuild("../hivemq-extension-sdk")` in `settings.gradle.kts`).
+
+NOTE: You can also change the path in the `settings.gradle.kts` to the path of your `hivemq-extension-sdk`.
+
+Otherwise, it will not be possible to publish develop snapshots to your local repository.

build.gradle.kts

@@ -91,7 +91,7 @@ tasks.withType<Jar>().configureEach {
 
 tasks.javadoc {
     title = "${metadata.readableName} ${project.version} API"
-
+    isFailOnError = false
     doLast { // javadoc search fix for jdk 11 https://bugs.openjdk.java.net/browse/JDK-8215291
         copy {
             from("${buildDir}/docs/javadoc/search.js")

settings.gradle.kts

@@ -9,3 +9,9 @@ pluginManagement {
 }
 
 rootProject.name = "hivemq-kafka-extension-customization-sdk"
+
+if (file("../hivemq-extension-sdk").exists()) {
+    if (gradle.parent == null) { // not part of a composite build
+        includeBuild("../hivemq-extension-sdk")
+    }
+}

src/main/java/com/hivemq/extensions/kafka/api/model/KafkaHeader.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.hivemq.extensions.kafka.api.model;
 
 import com.hivemq.extension.sdk.api.annotations.DoNotImplement;
@@ -52,11 +53,11 @@ public interface KafkaHeader {
      * @return the value of this header.
      * @since 4.4.0
      */
-    @NotNull byte[] getValueAsByteArray();
+    byte @NotNull [] getValueAsByteArray();
 
     /**
-     * @return the value of this header as a string. {@link java.nio.charset.StandardCharsets#UTF_8} is used for the
-     *         decoding.
+     * @return the value of this header as a string. {@link java.nio.charset.StandardCharsets#UTF_8 UTF_8} is used for
+     *         the decoding.
      * @since 4.4.0
      */
     @NotNull String getValueAsString();

src/main/java/com/hivemq/extensions/kafka/api/services/KafkaTopicService.java

@@ -91,7 +91,7 @@ public interface KafkaTopicService {
      * @throws NullPointerException if {@code topics} is or contains null.
      * @since 4.4.0
      */
-    @NotNull Map<String, @NotNull KafkaTopicState> createKafkaTopics(@NotNull Set<@NotNull String> topics);
+    @Immutable @NotNull Map<String, @NotNull KafkaTopicState> createKafkaTopics(@NotNull Set<@NotNull String> topics);
 
     /**
      * KafkaTopicState encodes the current known state of a Kafka topic on the associated cluster.

src/main/java/com/hivemq/extensions/kafka/api/transformers/TransformerInitInput.java

@@ -16,8 +16,10 @@
 
 package com.hivemq.extensions.kafka.api.transformers;
 
+import com.codahale.metrics.MetricRegistry;
 import com.hivemq.extension.sdk.api.annotations.DoNotImplement;
 import com.hivemq.extension.sdk.api.annotations.Immutable;
+import com.hivemq.extension.sdk.api.annotations.NotNull;
 
 /**
  * A marker interface for the input object of the {@link Transformer#init(TransformerInitInput)} method.
@@ -28,4 +30,15 @@
  */
 @Immutable
 @DoNotImplement
-public interface TransformerInitInput {}
+public interface TransformerInitInput {
+
+    /**
+     * Get the {@link MetricRegistry} of this HiveMQ node. It is possible to add own metrics to monitor
+     * custom business logic.
+     *
+     * @return the {@link MetricRegistry} of the HiveMQ node this Enterprise Extension for Kafka is running on.
+     * @since 4.5.0
+     */
+    @NotNull MetricRegistry getMetricRegistry();
+
+}

src/main/java/com/hivemq/extensions/kafka/api/transformers/kafkatomqtt/KafkaToMqttInitInput.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2020-present HiveMQ GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hivemq.extensions.kafka.api.transformers.kafkatomqtt;
+
+import com.hivemq.extension.sdk.api.annotations.DoNotImplement;
+import com.hivemq.extension.sdk.api.annotations.Immutable;
+import com.hivemq.extensions.kafka.api.transformers.TransformerInitInput;
+
+/**
+ * Provides context for the set up of a {@link KafkaToMqttTransformer}.
+ *
+ * @author Georg Held
+ * @since 4.5.0
+ */
+@Immutable
+@DoNotImplement
+public interface KafkaToMqttInitInput extends TransformerInitInput {}

src/main/java/com/hivemq/extensions/kafka/api/transformers/kafkatomqtt/KafkaToMqttInput.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2020-present HiveMQ GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hivemq.extensions.kafka.api.transformers.kafkatomqtt;
+
+import com.hivemq.extension.sdk.api.annotations.DoNotImplement;
+import com.hivemq.extension.sdk.api.annotations.Immutable;
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+import com.hivemq.extensions.kafka.api.model.KafkaCluster;
+import com.hivemq.extensions.kafka.api.model.KafkaRecord;
+
+/**
+ * The input parameter of the {@link KafkaToMqttTransformer}. It contains the information of the to be transformed
+ * {@link KafkaRecord}.
+ * <p>
+ * The MqttToKafkaInput allows access to the {@link KafkaCluster}.
+ *
+ * @author Christoph Schäbel
+ * @author Georg Held
+ * @since 4.5.0
+ */
+@Immutable
+@DoNotImplement
+public interface KafkaToMqttInput {
+
+    /**
+     * @return the {@link KafkaRecord} that triggered this transformer call.
+     * @since 4.5.0
+     */
+    @NotNull KafkaRecord getKafkaRecord();
+
+    /**
+     * @return the {@link KafkaCluster} the transformer is associated with.
+     * @since 4.5.0
+     */
+    @NotNull KafkaCluster getKafkaCluster();
+}

src/main/java/com/hivemq/extensions/kafka/api/transformers/kafkatomqtt/KafkaToMqttOutput.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2020-present HiveMQ GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hivemq.extensions.kafka.api.transformers.kafkatomqtt;
+
+import com.hivemq.extension.sdk.api.annotations.DoNotImplement;
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+import com.hivemq.extension.sdk.api.services.builder.PublishBuilder;
+import com.hivemq.extension.sdk.api.services.publish.Publish;
+
+import java.util.List;
+
+/**
+ * The output parameter of the {@link KafkaToMqttTransformer}. It allows access to the {@link PublishBuilder}.
+ * <p>
+ * After the {@link KafkaToMqttTransformer#transformKafkaToMqtt(KafkaToMqttInput, KafkaToMqttOutput)} method returns the
+ * {@link Publish}es given to this output will be published by HiveMQ.
+ *
+ * @author Christoph Schäbel
+ * @author Georg Held
+ * @since 4.5.0
+ */
+@DoNotImplement
+public interface KafkaToMqttOutput {
+
+    /**
+     * @return a new {@link PublishBuilder}.
+     * @since 4.5.0
+     */
+    @NotNull PublishBuilder newPublishBuilder();
+
+    /**
+     * Sets the {@link Publish}es, that will be published by HiveMQ after the {@link
+     * KafkaToMqttTransformer#transformKafkaToMqtt(KafkaToMqttInput, KafkaToMqttOutput)} call returns. The HiveMQ
+     * Enterprise Extension for Kafka will publish these publishes in the order provided by the {@code publishes}
+     * argument.
+     * <p>
+     * If desired, the same publish can occupy multiple places in the {@code publishes} list. When no publish shall be
+     * published by HiveMQ for a {@link com.hivemq.extensions.kafka.api.model.KafkaRecord}, call this method with an
+     * empty list.
+     * <p>
+     * Use the {@link #newPublishBuilder() PublishBuilder} to create new publishes as desired.
+     * <p>
+     * Each additional call of this method will overwrite the previous one.
+     *
+     * @param publishes a list of to be published {@link Publish}es.
+     * @throws NullPointerException     if {@code publishes} or any element of it is null.
+     * @throws IllegalArgumentException if any element in {@code publishes} was not created via a {@link
+     *                                  PublishBuilder}.
+     * @since 4.5.0
+     */
+    void setPublishes(@NotNull List<@NotNull Publish> publishes);
+}

src/main/java/com/hivemq/extensions/kafka/api/transformers/kafkatomqtt/KafkaToMqttTransformer.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2020-present HiveMQ GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hivemq.extensions.kafka.api.transformers.kafkatomqtt;
+
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+import com.hivemq.extension.sdk.api.annotations.ThreadSafe;
+import com.hivemq.extensions.kafka.api.transformers.Transformer;
+
+/**
+ * Implement this transformer for the programmatic creation of {@link com.hivemq.extension.sdk.api.services.publish.Publish
+ * Publishes} from {@link com.hivemq.extensions.kafka.api.model.KafkaRecord KafkaRecords}. One instance of the
+ * implementing class is created per reference in the kafka-configuration.xml. The methods of this interface may be
+ * called concurrently and must be thread-safe.
+ * <p>
+ * Your implementation of the KafkaToMqttTransformer must be placed in a Java archive (.jar) together with all its
+ * dependencies in the {@code customizations} folder of the HiveMQ Enterprise Extension for Kafka. In addition a {@code
+ * <kafka-to-mqtt-transformer>} referencing the implementing class via its canonical name must be configured in the
+ * {@code kafka-extension.xml} file.
+ *
+ * @author Christoph Schäbel
+ * @author Georg Held
+ * @author Daniel Krüger
+ * @since 4.5.0
+ */
+@FunctionalInterface
+public interface KafkaToMqttTransformer extends Transformer<KafkaToMqttInitInput> {
+
+    /**
+     * This callback is executed for every {@link com.hivemq.extensions.kafka.api.model.KafkaRecord KafkaRecord} that is
+     * polled by the HiveMQ Enterprise Extension for Kafka and matches the {@code <mqtt-to-kafka-transformer>} tag
+     * configured in the {@code <mqtt-topic-filters>}. It allows the publication of any number of {@link
+     * com.hivemq.extension.sdk.api.services.publish.Publish Publishes} via the {@link KafkaToMqttOutput} object. This
+     * method is called by multiple threads concurrently. Extensions are responsible for their own exception handling
+     * and this method must not throw any {@link Exception}.
+     *
+     * @param input  the {@link KafkaToMqttInput} contains the triggering {@link com.hivemq.extensions.kafka.api.model.KafkaRecord
+     *               KafkaRecord} and the {@link com.hivemq.extensions.kafka.api.model.KafkaCluster KafkaCluster}
+     *               information.
+     * @param output the {@link KafkaToMqttOutput} allows to {@link KafkaToMqttOutput#setPublishes(java.util.List)
+     *               provide a list of Publishes}. If no output is set, a empty List is used as default and the kafka
+     *               record will not be processed again, but ignored.
+     * @since 4.5.0
+     */
+    @ThreadSafe
+    void transformKafkaToMqtt(@NotNull KafkaToMqttInput input, @NotNull KafkaToMqttOutput output);
+}