pom.xml refactor

Author: uts58

README.md

@@ -20,11 +20,11 @@ To build the connector, you must have the following installed:
 
 Clone the repository with the following command:
 ```
-git clone https://github.com/uts58/kafka-connect-mqtt.git
+git clone https://github.com/uts58/kafka-connect-mqtt-file.git
 ```
-Change directory into the `kafka-connect-mqtt` directory
+Change directory into the `kafka-connect-mqtt-file` directory
 ```
-cd kafka-connect-mqtt
+cd kafka-connect-mqtt-file
 ```
 Build the connector fat jar using Maven
 ```
@@ -49,11 +49,6 @@ http://<kafkaconnect>:8083/connector-plugins
 If you see these entries, the connector has been installed succesfully
 
 ```
-{
-    "class": "org.ndsu.agda.connect.connectors.mqtt.MQTTSinkConnector",
-    "type": "sink",
-    "version": "<version>"
-},
 {
     "class": "org.ndsu.agda.connect.connectors.mqtt.MQTTSourceConnector",
     "type": "source",
@@ -99,42 +94,6 @@ curl -X POST \
 * `mqtt.userName` (optional): Username to connect to MQTT broker
 * `mqtt.password` (optional): Password to connect to MQTT broker
 
-## Configuring the MQTT Sink connector
-
-The MQTT Sink Connector reads messages from a Kafka topic and publishes them to a MQTT topic.
-
-Here is a basic configuration example:
-```
-curl -X POST \
-  http://<kafkaconnect>>:8083/connectors \
-  -H 'Content-Type: application/json' \
-  -d '{ "name": "mqtt-sink-connector",
-    "config":
-    {
-      "connector.class":"org.ndsu.agda.connect.connectors.mqtt.MQTTSinkConnector",
-      "mqtt.topic":"my_mqtt_topic",
-      "topics":"my_kafka_topic",
-      "mqtt.qos": 2,
-      "mqtt.clientID":"my_client_id",
-      "mqtt.broker":"tcp://127.0.0.1:1883",
-      "key.converter":"org.apache.kafka.connect.storage.StringConverter",
-      "key.converter.schemas.enable":false,
-      "value.converter":"org.apache.kafka.connect.storage.StringConverter",
-      "value.converter.schemas.enable":false
-    }
-}'
-```
-
-### Optional Configuration options
-* `mqtt.qos` (optional): 0 – At most Once, 1 – At Least Once, 2 – Exactly Once
-* `mqtt.automaticReconnect` (optional)(default: true): Should the client automatically reconnect in case of connection failures
-* `mqtt.keepAliveInterval` (optional)(default: 60 seconds)
-* `mqtt.cleanSession` (optional)(default: true): Controls the state after disconnecting the client from the broker.
-* `mqtt.connectionTimeout` (optional)(default: 30 seconds)
-* `mqtt.userName` (optional): Username to connect to MQTT broker
-* `mqtt.password` (optional): Password to connect to MQTT broker
-
-
 ## Configuring the File Sink connector
 
 The File Sink Connector reads messages from a Kafka topic(s) and dumps them into files.

pom.xml

@@ -7,7 +7,7 @@
 
     <groupId>org.ndsu.adga</groupId>
     <artifactId>kafka-connect</artifactId>
-    <version>1.2.2</version>
+    <version>1.2.3</version>
     <packaging>jar</packaging>
     <name>kafka.connect</name>
 
@@ -46,17 +46,11 @@
             <version>33.0.0-jre</version>
         </dependency>
 
-            <dependency>
-                <groupId>org.mongodb</groupId>
-                <artifactId>bson</artifactId>
-                <version>5.3.0</version>
-            </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
             <version>2.18.2</version>
         </dependency>
-
     </dependencies>
 
     <build>
@@ -115,4 +109,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

src/main/java/org/ndsu/agda/connect/connectors/file/FileSinkJsonWriter.java

@@ -1,6 +1,5 @@
 package org.ndsu.agda.connect.connectors.file;
 
-import org.apache.kafka.connect.errors.ConnectException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

src/main/java/org/ndsu/agda/connect/connectors/mqtt/MQTTSourceConverter.java

@@ -9,6 +9,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -35,14 +36,16 @@ protected SourceRecord convert(String topic, MqttMessage mqttMessage) {
         try {
             jsonMap = OBJECT_MAPPER.readValue(payload, new TypeReference<Map<String, Object>>() {
             });
+            jsonMap.put("mqttReceivedAt", Instant.now().toString());
+
             if (jsonMap.containsKey("iotnode")) {
                 @SuppressWarnings("unchecked")
                 Map<String, Object> iotNode = (Map<String, Object>) jsonMap.get("iotnode");
                 if (iotNode.containsKey("reportedAt")) {
-                    String reportedAtStr = (String) iotNode.get("reportedAt");
-                    jsonMap.put("reportedAt", reportedAtStr);
+                    jsonMap.put("baseReportedAt", iotNode.get("reportedAt").toString());
                 }
             }
+
         } catch (IOException e) {
             log.error("Failed to parse JSON payload, passing raw payload in a wrapper", e);
             jsonMap = Collections.singletonMap("raw", payload);