diff --git a/pom.xml b/pom.xml
index 2c65049..6f02890 100755
--- a/pom.xml
+++ b/pom.xml
@@ -50,14 +50,9 @@
             <scope>provided</scope>
         </dependency>
         <dependency>
-            <groupId>com.fasterxml.jackson.datatype</groupId>
-            <artifactId>jackson-datatype-jdk8</artifactId>
-            <version>2.9.9</version>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.datatype</groupId>
-            <artifactId>jackson-datatype-jsr310</artifactId>
-            <version>2.9.8</version>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-shaded-jackson</artifactId>
+            <version>2.10.1-10.0</version>
         </dependency>
         <dependency>
             <groupId>com.github.docker-java</groupId>
diff --git a/src/main/java/com/datastax/mcac/UnixSocketClient.java b/src/main/java/com/datastax/mcac/UnixSocketClient.java
index a9474e9..e87c1b8 100755
--- a/src/main/java/com/datastax/mcac/UnixSocketClient.java
+++ b/src/main/java/com/datastax/mcac/UnixSocketClient.java
@@ -906,7 +906,7 @@ private int writeMetric(String name, String tags, long[] estimatedBuckets)
             com.datastax.mcac.insights.metrics.Histogram h = new com.datastax.mcac.insights.metrics.Histogram(
                     name,
                     System.currentTimeMillis(),
-                    Optional.of(globalTags),
+                    globalTags,
                     hist.count(),
                     new SamplingStats(min, max, mean, p50, p75, p90, p98, p99, p999, stddev)
             );
@@ -965,7 +965,7 @@ private int writeMetric(String name, String tags, Timer timer)
             com.datastax.mcac.insights.metrics.Timer t = new com.datastax.mcac.insights.metrics.Timer(
                     name,
                     System.currentTimeMillis(),
-                    Optional.of(buckets),
+                    buckets,
                     timer.getCount(),
                     new SamplingStats((long) min, (long) max, mean, p50, p75, p95, p98, p99, p999, stddev),
                     new RateStats(min1Rate, min5rate, min15rate, meanRate),
@@ -1005,7 +1005,7 @@ private int writeMetric(String name, String tags, Meter meter)
             com.datastax.mcac.insights.metrics.Meter m = new com.datastax.mcac.insights.metrics.Meter(
                     name,
                     System.currentTimeMillis(),
-                    Optional.of(globalTags),
+                    globalTags,
                     meter.getCount(),
                     new RateStats(min1Rate, min5rate, min15rate, meanRate),
                     rateUnit.name()
@@ -1053,7 +1053,7 @@ private int writeMetric(String name, String tags, Histogram histogram)
             com.datastax.mcac.insights.metrics.Histogram h = new com.datastax.mcac.insights.metrics.Histogram(
                     name,
                     System.currentTimeMillis(),
-                    Optional.of(globalTags),
+                    globalTags,
                     histogram.getCount(),
                     new SamplingStats(min, max, mean, p50, p75, p95, p98, p99, p999, stddev)
             );
diff --git a/src/main/java/com/datastax/mcac/insights/Insight.java b/src/main/java/com/datastax/mcac/insights/Insight.java
index c63c40f..8dc6d4e 100755
--- a/src/main/java/com/datastax/mcac/insights/Insight.java
+++ b/src/main/java/com/datastax/mcac/insights/Insight.java
@@ -3,8 +3,8 @@
 import java.util.Objects;
 
 import com.datastax.mcac.utils.JacksonUtil;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 /**
  * Instead of providing both `Object o` and `InsightsMetadata meta` to the associated <br>
diff --git a/src/main/java/com/datastax/mcac/insights/InsightMetadata.java b/src/main/java/com/datastax/mcac/insights/InsightMetadata.java
index 2ce951f..e610e68 100755
--- a/src/main/java/com/datastax/mcac/insights/InsightMetadata.java
+++ b/src/main/java/com/datastax/mcac/insights/InsightMetadata.java
@@ -1,7 +1,7 @@
 package com.datastax.mcac.insights;
 
 import java.time.Instant;
-import java.util.HashMap;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -9,8 +9,9 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 public class InsightMetadata
 {
@@ -29,13 +30,14 @@ public enum InsightType
     @JsonProperty("name")
     public final String name;
     @JsonProperty("timestamp")
-    public final Optional<Long> timestamp;
+    public final Long timestamp;
     @JsonProperty("tags")
     public final Map<String, String> tags;
     @JsonProperty("insightType")
-    public final Optional<InsightType> insightType;
+    public final InsightType insightType;
     @JsonProperty("insightMappingId")
-    public Optional<String> insightMappingId;
+    @JsonInclude(JsonInclude.Include.NON_NULL) //ignore null field on this property only
+    public String insightMappingId;
 
     /**
      * InsightsMetadata base constructor which all other constructors map to.
@@ -57,13 +59,13 @@ public InsightMetadata(
             @JsonProperty("name")
                     String name,
             @JsonProperty("timestamp")
-                    Optional<Long> timestamp,
+                    Long timestamp,
             @JsonProperty("tags")
-                    Optional<Map<String, String>> tags,
+                    Map<String, String> tags,
             @JsonProperty("insightType")
-                    Optional<InsightType> insightType,
+                    InsightType insightType,
             @JsonProperty("insightMappingId")
-                    Optional<String> insightMappingId
+                    String insightMappingId
     )
     {
         /*
@@ -78,9 +80,9 @@ public InsightMetadata(
         Preconditions.checkArgument(!Strings.isNullOrEmpty(name), "name is required");
 
         this.name = name;
-        this.timestamp = Optional.of(timestamp.filter(ts -> ts > 0).orElse(Instant.now().toEpochMilli()));
-        this.tags = tags.orElseGet(HashMap::new); //kept mutable since we want to add some default tags
-        this.insightType = Optional.of(insightType.orElse(InsightType.EVENT));
+        this.timestamp = timestamp == null || timestamp <= 0L ? Instant.now().toEpochMilli() : timestamp;
+        this.tags = tags == null ? Collections.emptyMap() : tags;  //kept mutable since we want to add some default tags
+        this.insightType = insightType == null ? InsightType.EVENT : insightType;
         this.insightMappingId = insightMappingId;
     }
 
@@ -99,7 +101,7 @@ public InsightMetadata(
             Map<String, String> tags
     )
     {
-        this(name, Optional.ofNullable(timestamp), Optional.ofNullable(tags), Optional.empty(), Optional.empty());
+        this(name, timestamp, tags, null, null);
     }
 
     /**
@@ -115,7 +117,7 @@ public InsightMetadata(
             Map<String, String> tags
     )
     {
-        this(name, Optional.empty(), Optional.ofNullable(tags), Optional.empty(), Optional.empty());
+        this(name, null, tags, null, null);
     }
 
     /**
@@ -130,7 +132,7 @@ public InsightMetadata(
             Long timestamp
     )
     {
-        this(name, Optional.ofNullable(timestamp), Optional.empty(), Optional.empty(), Optional.empty());
+        this(name, timestamp, null, null, null);
     }
 
     /**
@@ -141,17 +143,17 @@ public InsightMetadata(
      */
     public InsightMetadata(String name)
     {
-        this(name, Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
+        this(name, null, null, null, null);
     }
 
     public InsightMetadata withTags(Map<String, String> newTags)
     {
-        return new InsightMetadata(this.name, this.timestamp, Optional.of(newTags), insightType, insightMappingId);
+        return new InsightMetadata(this.name, this.timestamp, newTags, insightType, insightMappingId);
     }
 
     public InsightMetadata withName(String newName)
     {
-        return new InsightMetadata(newName, this.timestamp, Optional.of(tags), insightType, insightMappingId);
+        return new InsightMetadata(newName, this.timestamp, tags, insightType, insightMappingId);
     }
 
     @Override
diff --git a/src/main/java/com/datastax/mcac/insights/events/ClientConnectionInformation.java b/src/main/java/com/datastax/mcac/insights/events/ClientConnectionInformation.java
index 80d296a..9b5f5a5 100755
--- a/src/main/java/com/datastax/mcac/insights/events/ClientConnectionInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/ClientConnectionInformation.java
@@ -29,12 +29,7 @@ public class ClientConnectionInformation extends Insight
 
     public ClientConnectionInformation(UUID sessionId, ClientState clientState, Map<String, String> options, boolean isHeartbeat)
     {
-        super(new InsightMetadata(
-                        isHeartbeat ? NAME_HEARTBEAT : NAME,
-                        Optional.of(System.currentTimeMillis()),
-                        Optional.empty(),
-                        Optional.of(InsightMetadata.InsightType.EVENT),
-                        Optional.empty()),
+        super(new InsightMetadata(isHeartbeat ? NAME_HEARTBEAT : NAME),
                 new Data(sessionId,  clientState, options));
     }
 
diff --git a/src/main/java/com/datastax/mcac/insights/events/CompactionEndedInformation.java b/src/main/java/com/datastax/mcac/insights/events/CompactionEndedInformation.java
index 83c4226..b7f756d 100755
--- a/src/main/java/com/datastax/mcac/insights/events/CompactionEndedInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/CompactionEndedInformation.java
@@ -6,10 +6,10 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 @JsonIgnoreProperties(ignoreUnknown = true)
 public final class CompactionEndedInformation extends Insight
@@ -32,10 +32,10 @@ public CompactionEndedInformation(
     {
         super(new InsightMetadata(
                         NAME,
-                        Optional.of(System.currentTimeMillis()),
-                        Optional.empty(),
-                        Optional.of(InsightMetadata.InsightType.EVENT),
-                        Optional.of("version1")
+                        System.currentTimeMillis(),
+                        null,
+                        InsightMetadata.InsightType.EVENT,
+                        "version1"
                 ),
                 new DataVersion1(
                         id,
diff --git a/src/main/java/com/datastax/mcac/insights/events/CompactionStartedInformation.java b/src/main/java/com/datastax/mcac/insights/events/CompactionStartedInformation.java
index 88bf942..1567707 100755
--- a/src/main/java/com/datastax/mcac/insights/events/CompactionStartedInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/CompactionStartedInformation.java
@@ -6,10 +6,10 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class CompactionStartedInformation extends Insight
@@ -30,10 +30,10 @@ public CompactionStartedInformation(
     {
         super(new InsightMetadata(
                         NAME,
-                        Optional.of(System.currentTimeMillis()),
-                        Optional.empty(),
-                        Optional.of(InsightMetadata.InsightType.EVENT),
-                        Optional.of("version1")
+                        System.currentTimeMillis(),
+                        null,
+                        InsightMetadata.InsightType.EVENT,
+                        "version1"
                 ),
                 new CompactionStartedInformation.DataVersion1(
                         id,
diff --git a/src/main/java/com/datastax/mcac/insights/events/DroppedMessageInformation.java b/src/main/java/com/datastax/mcac/insights/events/DroppedMessageInformation.java
index c53a8b8..b3c7667 100755
--- a/src/main/java/com/datastax/mcac/insights/events/DroppedMessageInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/DroppedMessageInformation.java
@@ -2,7 +2,7 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.util.Optional;
 
@@ -20,13 +20,7 @@ public DroppedMessageInformation(
     )
     {
         super(
-                new InsightMetadata(
-                        NAME,
-                        Optional.of(System.currentTimeMillis()),
-                        Optional.empty(),
-                        Optional.of(InsightMetadata.InsightType.EVENT),
-                        Optional.empty()
-                ),
+                new InsightMetadata(NAME),
                 new Data(
                         groupName,
                         interval,
diff --git a/src/main/java/com/datastax/mcac/insights/events/ExceptionInformation.java b/src/main/java/com/datastax/mcac/insights/events/ExceptionInformation.java
index 149459d..7c1ee9c 100755
--- a/src/main/java/com/datastax/mcac/insights/events/ExceptionInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/ExceptionInformation.java
@@ -8,7 +8,7 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 public class ExceptionInformation extends Insight
 {
@@ -16,12 +16,7 @@ public class ExceptionInformation extends Insight
 
     public ExceptionInformation(Throwable t)
     {
-        super(new InsightMetadata(
-                        NAME,
-                        Optional.of(System.currentTimeMillis()),
-                        Optional.empty(),
-                        Optional.of(InsightMetadata.InsightType.EVENT),
-                        Optional.empty()),
+        super(new InsightMetadata(NAME),
                 new Data(t));
     }
 
diff --git a/src/main/java/com/datastax/mcac/insights/events/FlushInformation.java b/src/main/java/com/datastax/mcac/insights/events/FlushInformation.java
index e4c4c5c..0341c56 100755
--- a/src/main/java/com/datastax/mcac/insights/events/FlushInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/FlushInformation.java
@@ -1,14 +1,13 @@
 package com.datastax.mcac.insights.events;
 
 import java.util.Collection;
-import java.util.List;
 import java.util.Optional;
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.cassandra.db.Memtable;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 public class FlushInformation extends Insight
 {
@@ -16,13 +15,7 @@ public class FlushInformation extends Insight
 
     public FlushInformation(Memtable memtable, Collection<SSTableReader> sstables, long durationInMillis, boolean isTruncate)
     {
-        super(new InsightMetadata(
-                NAME,
-                Optional.of(System.currentTimeMillis()),
-                Optional.empty(),
-                Optional.of(InsightMetadata.InsightType.EVENT),
-                Optional.empty()
-        ), new Data(
+        super(new InsightMetadata(NAME), new Data(
                 memtable,
                 sstables,
                 durationInMillis,
diff --git a/src/main/java/com/datastax/mcac/insights/events/GCInformation.java b/src/main/java/com/datastax/mcac/insights/events/GCInformation.java
index 6fc0405..5577bf0 100755
--- a/src/main/java/com/datastax/mcac/insights/events/GCInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/GCInformation.java
@@ -7,9 +7,9 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 @JsonIgnoreProperties(ignoreUnknown = true)
 public final class GCInformation extends Insight
diff --git a/src/main/java/com/datastax/mcac/insights/events/GossipChangeInformation.java b/src/main/java/com/datastax/mcac/insights/events/GossipChangeInformation.java
index 6e744be..0551046 100755
--- a/src/main/java/com/datastax/mcac/insights/events/GossipChangeInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/GossipChangeInformation.java
@@ -1,21 +1,18 @@
 package com.datastax.mcac.insights.events;
 
 import java.net.InetAddress;
-import java.util.Collections;
 import java.util.EnumMap;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.cassandra.gms.ApplicationState;
 import org.apache.cassandra.gms.EndpointState;
 import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.gms.VersionedValue;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 public class GossipChangeInformation extends Insight
 {
@@ -23,13 +20,7 @@ public class GossipChangeInformation extends Insight
 
     public GossipChangeInformation(Data data)
     {
-        super(new InsightMetadata(
-                NAME,
-                Optional.of(System.currentTimeMillis()),
-                Optional.empty(),
-                Optional.of(InsightMetadata.InsightType.EVENT),
-                Optional.empty()
-        ), data);
+        super(new InsightMetadata(NAME), data);
     }
 
     public GossipChangeInformation(GossipEventType eventType, InetAddress ipAddress)
diff --git a/src/main/java/com/datastax/mcac/insights/events/InsightsClientStarted.java b/src/main/java/com/datastax/mcac/insights/events/InsightsClientStarted.java
index 6bedb7d..960c12e 100755
--- a/src/main/java/com/datastax/mcac/insights/events/InsightsClientStarted.java
+++ b/src/main/java/com/datastax/mcac/insights/events/InsightsClientStarted.java
@@ -2,7 +2,7 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 
 /*
  * Event published through the insights client to indicate
@@ -16,10 +16,7 @@ public class InsightsClientStarted extends Insight
     public InsightsClientStarted()
     {
         super(
-                new InsightMetadata(
-                        NAME,
-                        System.currentTimeMillis()
-                ),
+                new InsightMetadata(NAME),
                 null
         );
     }
diff --git a/src/main/java/com/datastax/mcac/insights/events/LargePartitionInformation.java b/src/main/java/com/datastax/mcac/insights/events/LargePartitionInformation.java
index 9a05abd..5b326df 100755
--- a/src/main/java/com/datastax/mcac/insights/events/LargePartitionInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/LargePartitionInformation.java
@@ -4,9 +4,9 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 public class LargePartitionInformation extends Insight
 {
@@ -14,12 +14,7 @@ public class LargePartitionInformation extends Insight
 
     public LargePartitionInformation(String keyspace, String table, DecoratedKey key, long size, long partitionLimit)
     {
-        super(new InsightMetadata(
-                        NAME,
-                        Optional.of(System.currentTimeMillis()),
-                        Optional.empty(),
-                        Optional.of(InsightMetadata.InsightType.EVENT),
-                        Optional.empty()),
+        super(new InsightMetadata(NAME),
                 new Data(keyspace, table,
                         key.getToken() instanceof Murmur3Partitioner.LongToken ? (Long) key.getToken().getTokenValue() : Long.MIN_VALUE,
                         size, partitionLimit));
diff --git a/src/main/java/com/datastax/mcac/insights/events/MCACFingerprint.java b/src/main/java/com/datastax/mcac/insights/events/MCACFingerprint.java
index 9f2f4d1..366d4ad 100755
--- a/src/main/java/com/datastax/mcac/insights/events/MCACFingerprint.java
+++ b/src/main/java/com/datastax/mcac/insights/events/MCACFingerprint.java
@@ -4,8 +4,8 @@
 
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 public class MCACFingerprint extends Insight
 {
@@ -25,14 +25,7 @@ public MCACFingerprint(
 
     public MCACFingerprint(List<OSSFingerprint> fingerprintList)
     {
-        super(new InsightMetadata(
-                NAME,
-                //TODO: if we upgrade c* dependency we could use ApproximateTime
-                Optional.of(System.currentTimeMillis()),
-                Optional.empty(),
-                Optional.of(InsightMetadata.InsightType.EVENT),
-                Optional.empty()
-        ), new Data(fingerprintList));
+        super(new InsightMetadata(NAME), new Data(fingerprintList));
     }
 
     public static class Data
diff --git a/src/main/java/com/datastax/mcac/insights/events/NodeConfiguration.java b/src/main/java/com/datastax/mcac/insights/events/NodeConfiguration.java
index 466db74..d2172c8 100755
--- a/src/main/java/com/datastax/mcac/insights/events/NodeConfiguration.java
+++ b/src/main/java/com/datastax/mcac/insights/events/NodeConfiguration.java
@@ -25,24 +25,24 @@
 import com.datastax.mcac.insights.InsightMetadata;
 import com.datastax.mcac.utils.JacksonUtil;
 import com.datastax.mcac.utils.ShellUtils;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonRawValue;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.ObjectWriter;
-import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
-import com.fasterxml.jackson.databind.ser.std.MapProperty;
 import org.apache.cassandra.config.Config;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.fasterxml.jackson.annotation.JsonFilter;
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.core.JsonStreamContext;
-import com.fasterxml.jackson.databind.SerializerProvider;
-import com.fasterxml.jackson.databind.ser.PropertyWriter;
-import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonFilter;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonRawValue;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonStreamContext;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectWriter;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.PropertyWriter;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.MapProperty;
 
 
 /**
@@ -74,10 +74,10 @@ public NodeConfiguration(Data data)
     {
         super(new InsightMetadata(
                 NAME,
-                Optional.of(System.currentTimeMillis()),
-                Optional.empty(),
-                Optional.of(InsightMetadata.InsightType.EVENT),
-                Optional.of(MAPPING_VERSION)
+                System.currentTimeMillis(),
+                null,
+                InsightMetadata.InsightType.EVENT,
+                MAPPING_VERSION
         ), data);
     }
 
@@ -236,7 +236,8 @@ private Data()
     @JsonFilter("secure filter")
     static class SecureFilterMixIn {}
 
-    static class SecurePropertyFilter extends SimpleBeanPropertyFilter {
+    static class SecurePropertyFilter extends SimpleBeanPropertyFilter
+    {
         private static final Pattern pattern = Pattern.compile("(secret|user|pass)", Pattern.CASE_INSENSITIVE);
 
         private String createPath(PropertyWriter writer, JsonGenerator jgen)
diff --git a/src/main/java/com/datastax/mcac/insights/events/NodeSystemInformation.java b/src/main/java/com/datastax/mcac/insights/events/NodeSystemInformation.java
index 6bbddfa..cd693fa 100755
--- a/src/main/java/com/datastax/mcac/insights/events/NodeSystemInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/NodeSystemInformation.java
@@ -14,10 +14,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonRawValue;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonRawValue;
 
 import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
 
@@ -25,10 +25,7 @@
 
 
 /**
- * Insight that holds all the meta-info we can get our hands on.
- *
- * This class is optimised for fast writes, not intended to be used
- * directly by DSE other than tests.
+ * Insight that holds all node meta-info
  */
 public class NodeSystemInformation extends Insight
 {
@@ -56,10 +53,10 @@ public NodeSystemInformation(Data data)
     {
         super(new InsightMetadata(
                 NAME,
-                Optional.of(System.currentTimeMillis()),
-                Optional.empty(),
-                Optional.of(InsightMetadata.InsightType.EVENT),
-                Optional.of(MAPPING_VERSION)
+                System.currentTimeMillis(),
+                null,
+                InsightMetadata.InsightType.EVENT,
+                MAPPING_VERSION
         ), data);
     }
 
diff --git a/src/main/java/com/datastax/mcac/insights/events/SSTableCompactionInformation.java b/src/main/java/com/datastax/mcac/insights/events/SSTableCompactionInformation.java
index 68bee90..afd0719 100755
--- a/src/main/java/com/datastax/mcac/insights/events/SSTableCompactionInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/SSTableCompactionInformation.java
@@ -1,6 +1,6 @@
 package com.datastax.mcac.insights.events;
 
-import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 public class SSTableCompactionInformation
 {
diff --git a/src/main/java/com/datastax/mcac/insights/events/SchemaInformation.java b/src/main/java/com/datastax/mcac/insights/events/SchemaInformation.java
index 646bb62..e24b9a4 100755
--- a/src/main/java/com/datastax/mcac/insights/events/SchemaInformation.java
+++ b/src/main/java/com/datastax/mcac/insights/events/SchemaInformation.java
@@ -3,14 +3,14 @@
 import com.datastax.mcac.insights.Insight;
 import com.datastax.mcac.insights.InsightMetadata;
 import com.datastax.mcac.utils.JacksonUtil;
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonRawValue;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonRawValue;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -20,11 +20,6 @@
 import java.util.Optional;
 
 
-/**
- * Copyright DataStax, Inc.
- *
- * Please see the included license file for details.
- */
 public class SchemaInformation extends Insight
 {
     public static final String NAME = "oss.insights.events.schema_information";
@@ -69,10 +64,10 @@ public SchemaInformation(Data data)
     {
         super(new InsightMetadata(
                 NAME,
-                Optional.of(System.currentTimeMillis()),
-                Optional.empty(),
-                Optional.of(InsightMetadata.InsightType.EVENT),
-                Optional.of(MAPPING_VERSION)
+                System.currentTimeMillis(),
+                null,
+                InsightMetadata.InsightType.EVENT,
+                MAPPING_VERSION
         ), data);
     }
 
diff --git a/src/main/java/com/datastax/mcac/insights/metrics/Counter.java b/src/main/java/com/datastax/mcac/insights/metrics/Counter.java
index ddd688a..17e33c9 100755
--- a/src/main/java/com/datastax/mcac/insights/metrics/Counter.java
+++ b/src/main/java/com/datastax/mcac/insights/metrics/Counter.java
@@ -12,8 +12,8 @@ public class Counter extends Metric
     @JsonCreator
     public Counter(
             @JsonProperty("name") String name,
-            @JsonProperty("timestamp") Optional<Long> timestamp,
-            @JsonProperty("tags") Optional<Map<String, String>> tags,
+            @JsonProperty("timestamp") Long timestamp,
+            @JsonProperty("tags") Map<String, String> tags,
             @JsonProperty("count") long count
     )
     {
diff --git a/src/main/java/com/datastax/mcac/insights/metrics/Gauge.java b/src/main/java/com/datastax/mcac/insights/metrics/Gauge.java
index ea1ff92..70c9bb8 100755
--- a/src/main/java/com/datastax/mcac/insights/metrics/Gauge.java
+++ b/src/main/java/com/datastax/mcac/insights/metrics/Gauge.java
@@ -13,8 +13,8 @@ public class Gauge extends Metric
     @JsonCreator
     public Gauge(
             @JsonProperty("name") String name,
-            @JsonProperty("timestamp") Optional<Long> timestamp,
-            @JsonProperty("tags") Optional<Map<String, String>> tags,
+            @JsonProperty("timestamp") Long timestamp,
+            @JsonProperty("tags") Map<String, String> tags,
             @JsonProperty("value") Number value
     )
     {
@@ -24,20 +24,10 @@ public Gauge(
     public Gauge(
             String name,
             Long timestamp,
-            Map<String, String> tags,
             Number value
     )
     {
-        super(name, Optional.of(timestamp), Optional.of(tags), InsightMetadata.InsightType.GAUGE, new Data(value));
-    }
-
-    public Gauge(
-            String name,
-            Long timestamp,
-            Number value
-    )
-    {
-        super(name, Optional.of(timestamp), Optional.empty(), InsightMetadata.InsightType.GAUGE, new Data(value));
+        super(name, timestamp, null, InsightMetadata.InsightType.GAUGE, new Data(value));
     }
 
     @JsonIgnore
diff --git a/src/main/java/com/datastax/mcac/insights/metrics/Histogram.java b/src/main/java/com/datastax/mcac/insights/metrics/Histogram.java
index 390b946..80f7f30 100755
--- a/src/main/java/com/datastax/mcac/insights/metrics/Histogram.java
+++ b/src/main/java/com/datastax/mcac/insights/metrics/Histogram.java
@@ -13,12 +13,12 @@ public class Histogram extends Metric
     public Histogram(
             @JsonProperty("name") String name,
             @JsonProperty("timestamp") Long timestamp,
-            @JsonProperty("tags") Optional<Map<String, String>> tags,
+            @JsonProperty("tags") Map<String, String> tags,
             @JsonProperty("count") long count,
             @JsonProperty("samplingStats") SamplingStats samplingStats
     )
     {
-        super(name, Optional.of(timestamp), tags, InsightMetadata.InsightType.HISTOGRAM, new Data(count, samplingStats));
+        super(name, timestamp, tags, InsightMetadata.InsightType.HISTOGRAM, new Data(count, samplingStats));
     }
 
     private static final class Data
diff --git a/src/main/java/com/datastax/mcac/insights/metrics/Meter.java b/src/main/java/com/datastax/mcac/insights/metrics/Meter.java
index 42a269a..da85755 100755
--- a/src/main/java/com/datastax/mcac/insights/metrics/Meter.java
+++ b/src/main/java/com/datastax/mcac/insights/metrics/Meter.java
@@ -13,7 +13,7 @@ public final class Meter extends Metric
     public Meter(
             @JsonProperty("name") String name,
             @JsonProperty("timestamp") Long timestamp,
-            @JsonProperty("tags") Optional<Map<String, String>> tags,
+            @JsonProperty("tags") Map<String, String> tags,
             @JsonProperty("count") long count,
             @JsonProperty("rateStats") RateStats rateStats,
             @JsonProperty("rateUnit") String rateUnit
@@ -21,7 +21,7 @@ public Meter(
     {
         super(
                 name,
-                Optional.of(timestamp),
+                timestamp,
                 tags,
                 InsightMetadata.InsightType.METER,
                 new Data(count, rateStats, rateUnit)
diff --git a/src/main/java/com/datastax/mcac/insights/metrics/Metric.java b/src/main/java/com/datastax/mcac/insights/metrics/Metric.java
index b3e933a..5026e1c 100755
--- a/src/main/java/com/datastax/mcac/insights/metrics/Metric.java
+++ b/src/main/java/com/datastax/mcac/insights/metrics/Metric.java
@@ -19,12 +19,12 @@ public enum MetricType
 
     public Metric(
             String name,
-            Optional<Long> timestamp,
-            Optional<Map<String, String>> tags,
+            Long timestamp,
+            Map<String, String> tags,
             InsightMetadata.InsightType insightType,
             Object metricData
     )
     {
-        super(new InsightMetadata(name, timestamp, tags, Optional.of(insightType), Optional.empty()), metricData);
+        super(new InsightMetadata(name, timestamp, tags, insightType, null), metricData);
     }
 }
diff --git a/src/main/java/com/datastax/mcac/insights/metrics/Timer.java b/src/main/java/com/datastax/mcac/insights/metrics/Timer.java
index 02a195e..bae39d6 100755
--- a/src/main/java/com/datastax/mcac/insights/metrics/Timer.java
+++ b/src/main/java/com/datastax/mcac/insights/metrics/Timer.java
@@ -13,7 +13,7 @@ public class Timer extends Metric
     public Timer(
             @JsonProperty("name") String name,
             @JsonProperty("timestamp") Long timestamp,
-            @JsonProperty("tags") Optional<Map<String, String>> tags,
+            @JsonProperty("tags") Map<String, String> tags,
             @JsonProperty("count") long count,
             @JsonProperty("samplingStats") SamplingStats samplingStats,
             @JsonProperty("rateStats") RateStats rateStats,
@@ -23,7 +23,7 @@ public Timer(
     {
         super(
                 name,
-                Optional.of(timestamp),
+                timestamp,
                 tags,
                 InsightMetadata.InsightType.TIMER,
                 new Data(count, samplingStats, rateStats, rateUnit, durationUnit)
diff --git a/src/main/java/com/datastax/mcac/utils/JacksonUtil.java b/src/main/java/com/datastax/mcac/utils/JacksonUtil.java
index 540539e..d129262 100755
--- a/src/main/java/com/datastax/mcac/utils/JacksonUtil.java
+++ b/src/main/java/com/datastax/mcac/utils/JacksonUtil.java
@@ -1,18 +1,17 @@
 package com.datastax.mcac.utils;
 
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.ObjectReader;
-import com.fasterxml.jackson.databind.ObjectWriter;
-import com.fasterxml.jackson.databind.SerializationFeature;
-import com.fasterxml.jackson.databind.Module;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.fasterxml.jackson.databind.util.ISO8601DateFormat;
-import com.fasterxml.jackson.databind.util.ISO8601Utils;
-import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
-import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.Module;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectReader;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectWriter;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializationFeature;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.util.ISO8601DateFormat;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.util.ISO8601Utils;
+
 import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableList;
@@ -41,8 +40,6 @@ public static class JacksonUtilException extends IOException
 
     private static final Logger logger = LoggerFactory.getLogger(JacksonUtil.class);
 
-    private static final ImmutableList<Module> modules = ImmutableList.of(new Jdk8Module(), new JavaTimeModule());
-
     private JacksonUtil()
     {
         // util class should have private constructor
@@ -50,22 +47,15 @@ private JacksonUtil()
 
     private static final Supplier<ObjectMapper> MAPPER_SUPPLIER = Suppliers.memoize(() -> {
         ObjectMapper mapper = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, true);
-        modules.forEach(mapper::registerModule);
         return mapper;
     });
 
 
     private static final Supplier<ObjectMapper> PRETTY_MAPPER_SUPPLIER = Suppliers.memoize(() -> {
         ObjectMapper mapper = new ObjectMapper().configure(SerializationFeature.INDENT_OUTPUT, true);
-        modules.forEach(mapper::registerModule);
         return mapper;
     });
-
-    public static List<Module> getObjectMapperModules()
-    {
-        return modules;
-    }
-
+    
     /**
      * <p>Public getters/setters/properties/creators will be auto detected.</p>
      * <p><b>NOTE</b>: public properties are allowed here to avoid getter/setter boilerplate