[ISSUE-114] Properly deserialize avro to json (#125)

* [ISSUE-114] Properly deserialize avro to json

* [ISSUE-114] update version, changelog

* add missing header

* Use updated commons-compress library

Author: Crim

CHANGELOG.md

@@ -2,6 +2,11 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## 2.1.2 (UNRELEASED)
+
+#### Bug fixes
+- [ISSUE-114](https://github.com/SourceLabOrg/kafka-webview/issues/114) Better out of the box support for Avro decoding / Confluent Schema Registry.
+
 ## 2.1.1 (01/08/2019)
 #### New Features
 - Added ability to Copy previously created views.

dev-cluster/pom.xml

@@ -5,12 +5,12 @@
     <parent>
         <artifactId>kafka-webview</artifactId>
         <groupId>org.sourcelab</groupId>
-        <version>2.1.1</version>
+        <version>2.1.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>dev-cluster</artifactId>
-    <version>2.1.1</version>
+    <version>2.1.2</version>
 
     <!-- Require Maven 3.3.9 -->
     <prerequisites>

kafka-webview-plugin/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.sourcelab</groupId>
         <artifactId>kafka-webview</artifactId>
-        <version>2.1.1</version>
+        <version>2.1.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>kafka-webview-plugin</artifactId>

kafka-webview-ui/pom.xml

@@ -5,11 +5,11 @@
     <parent>
         <artifactId>kafka-webview</artifactId>
         <groupId>org.sourcelab</groupId>
-        <version>2.1.1</version>
+        <version>2.1.2</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>kafka-webview-ui</artifactId>
-    <version>2.1.1</version>
+    <version>2.1.2</version>
 
     <!-- Module Description and Ownership -->
     <name>Kafka WebView UI</name>
@@ -30,9 +30,12 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <java.version>1.8</java.version>
+
+        <avro.version>1.8.2</avro.version>
         <bootstrap.version>4.0.0-beta</bootstrap.version>
-        <thymeleaf.version>3.0.9.RELEASE</thymeleaf.version>
         <kafka.version>2.0.0</kafka.version>
+        <protobuf.version>3.6.1</protobuf.version>
+        <thymeleaf.version>3.0.9.RELEASE</thymeleaf.version>
     </properties>
 
     <dependencies>
@@ -175,7 +178,27 @@
         <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
-            <version>3.6.1</version>
+            <version>${protobuf.version}</version>
+        </dependency>
+
+        <!-- Jackson support for avro-->
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>${avro.version}</version>
+            <exclusions>
+                <!-- exclude vulnerable dependency -->
+                <exclusion>
+                    <groupId>org.apache.commons</groupId>
+                    <artifactId>commons-compress</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <!-- Explicitly include updated version -->
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-compress</artifactId>
+            <version>1.18</version>
         </dependency>
 
         <!-- For tests -->

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/configuration/AppProperties.java

@@ -52,6 +52,9 @@ public class AppProperties {
     @Value("${app.requireSsl:false}")
     private boolean requireSsl = false;
 
+    @Value("${app.avro.includeSchema:true}")
+    private boolean avroIncludeSchema = true;
+
     /**
      * Flag read from configuration file to determine if the app should
      * enforce User authentication.
@@ -97,6 +100,10 @@ public LdapAppProperties getLdapProperties() {
         return ldapProperties;
     }
 
+    public boolean isAvroIncludeSchema() {
+        return avroIncludeSchema;
+    }
+
     @Override
     public String toString() {
         return "AppProperties{"

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/configuration/WebConfig.java

@@ -26,6 +26,9 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.hubspot.jackson.datatype.protobuf.ProtobufModule;
+import org.apache.avro.generic.GenericData;
+import org.sourcelab.kafka.webview.ui.manager.jackson.SimpleAvroDataSerializer;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
@@ -42,6 +45,9 @@
 @Configuration
 @EnableWebMvc
 public class WebConfig implements WebMvcConfigurer {
+    @Autowired
+    private AppProperties appProperties;
+
     @Override
     public void addResourceHandlers(ResourceHandlerRegistry registry) {
         // css resource
@@ -76,6 +82,7 @@ public void configureMessageConverters(List<HttpMessageConverter<?>> converters)
         final ObjectMapper mapper = Jackson2ObjectMapperBuilder
             .json()
             .modulesToInstall(new ProtobufModule())
+            .serializerByType(GenericData.Record.class, new SimpleAvroDataSerializer(appProperties.isAvroIncludeSchema()))
             .build();
         converters.add(new MappingJackson2HttpMessageConverter(mapper));
     }

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/jackson/SimpleAvroDataSerializer.java

@@ -0,0 +1,83 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2017, 2018, 2019 SourceLab.org (https://github.com/SourceLabOrg/kafka-webview/)
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.sourcelab.kafka.webview.ui.manager.jackson;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.avro.generic.GenericData;
+
+import java.io.IOException;
+
+/**
+ * Simple Avro to JSON serializer for Jackson.
+ */
+public class SimpleAvroDataSerializer extends JsonSerializer<GenericData.Record> {
+    private final boolean includeSchema;
+
+    /**
+     * No arg constructor.  Default to including the schema.
+     */
+    public SimpleAvroDataSerializer() {
+        this(true);
+    }
+
+    /**
+     * Constructor.
+     * @param includeSchema Should the schema be included.
+     */
+    public SimpleAvroDataSerializer(final boolean includeSchema) {
+        this.includeSchema = includeSchema;
+    }
+
+    @Override
+    public void serialize(final GenericData.Record value, final JsonGenerator gen, final SerializerProvider provider) throws IOException {
+        if (includeSchema) {
+            writeIncludingSchema(value, gen);
+        } else {
+            writeValueOnly(value, gen);
+        }
+    }
+
+    private void writeIncludingSchema(final GenericData.Record value, final JsonGenerator gen) throws IOException {
+        // Start new object.
+        gen.writeStartObject();
+
+        // Write value
+        gen.writeFieldName("value");
+        gen.writeRawValue(value.toString());
+
+        // Write schema
+        gen.writeFieldName("schema");
+        gen.writeRawValue(value.getSchema().toString());
+
+        // End object
+        gen.writeEndObject();
+    }
+
+    private void writeValueOnly(final GenericData.Record value, final JsonGenerator gen) throws IOException {
+        gen.writeRawValue(value.toString());
+    }
+}

pom.xml

@@ -7,7 +7,7 @@
     <groupId>org.sourcelab</groupId>
     <artifactId>kafka-webview</artifactId>
     <packaging>pom</packaging>
-    <version>2.1.1</version>
+    <version>2.1.2</version>
 
     <!-- Define submodules -->
     <modules>