[ISSUE-102] Start putting together a web publisher

kafka-webview-plugin/pom.xml

@@ -9,7 +9,7 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>kafka-webview-plugin</artifactId>
-    <version>1.0.0</version>
+    <version>2.0.0</version>
 
     <!-- Module Description and Ownership -->
     <name>Kafka WebView Plugin</name>
@@ -32,6 +32,12 @@
     </properties>
 
     <dependencies>
-        <!-- None! -->
+        <!-- Kafka Dependency Provided -->
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>${kafka.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 </project>

kafka-webview-plugin/src/main/java/org/sourcelab/kafka/webview/ui/plugin/serializer/SerializerTransformer.java

@@ -0,0 +1,68 @@
+/**
+ * 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.plugin.serializer;
+
+
+import org.apache.kafka.common.serialization.Serializer;
+
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * For translation of values input from Kafka WebView UI/Browser, to a Kafka serializer.
+ *
+ * Typically an implementation of this interface is only needed in situations where you want to serialize
+ * complex/composite objects which require multiple input values from the browser interface.
+ *
+ * @param <T> Object type the Kafka Serializer instance will be serializing.
+ */
+public interface SerializerTransformer<T> {
+    /**
+     * Configure this class.
+     * @param configs configs in key/value pairs
+     * @param isKey whether is for key or value
+     */
+    void configure(final Map<String, ?> configs, boolean isKey);
+
+    /**
+     * Transformation logic.
+     * @param topic The topic being produced to.
+     * @param valueMap Map of values to produce.
+     * @return Serialized/flattened value that will get passed to the serializer instance.
+     */
+    T transform(final String topic, final Map<String, String> valueMap);
+
+    /**
+     * Underlying Kafka value serializer class.
+     * @return Underlying Kafka value serializer class.
+     */
+    Class<? extends Serializer> getSerializerClass();
+
+    /**
+     * Return collection of field names to collect values from the web interface.
+     * @return Collection of file names.
+     */
+    Collection<String> getFieldNames();
+}

kafka-webview-ui/pom.xml

@@ -44,7 +44,7 @@
         <dependency>
             <groupId>org.sourcelab</groupId>
             <artifactId>kafka-webview-plugin</artifactId>
-            <version>1.0.0</version>
+            <version>2.0.0</version>
         </dependency>
 
         <!-- Kafka -->

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

@@ -24,6 +24,7 @@
 
 package org.sourcelab.kafka.webview.ui.configuration;
 
+import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.BytesDeserializer;
 import org.apache.kafka.common.serialization.DoubleDeserializer;
@@ -32,11 +33,17 @@
 import org.apache.kafka.common.serialization.LongDeserializer;
 import org.apache.kafka.common.serialization.ShortDeserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.sourcelab.kafka.webview.ui.manager.kafka.producer.transformer.LongTransformer;
+import org.sourcelab.kafka.webview.ui.manager.kafka.producer.transformer.StringTransformer;
 import org.sourcelab.kafka.webview.ui.manager.kafka.deserializer.BytesToHexDeserializer;
 import org.sourcelab.kafka.webview.ui.manager.user.UserManager;
 import org.sourcelab.kafka.webview.ui.model.MessageFormat;
+import org.sourcelab.kafka.webview.ui.model.PartitioningStrategy;
+import org.sourcelab.kafka.webview.ui.model.SerializerFormat;
 import org.sourcelab.kafka.webview.ui.model.UserRole;
 import org.sourcelab.kafka.webview.ui.repository.MessageFormatRepository;
+import org.sourcelab.kafka.webview.ui.repository.PartitioningStrategyRepository;
+import org.sourcelab.kafka.webview.ui.repository.SerializerFormatRepository;
 import org.sourcelab.kafka.webview.ui.repository.UserRepository;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.ApplicationArguments;
@@ -54,14 +61,22 @@ public final class DataLoaderConfig implements ApplicationRunner {
 
     private final MessageFormatRepository messageFormatRepository;
     private final UserRepository userRepository;
+    private final PartitioningStrategyRepository partitioningStrategyRepository;
+    private final SerializerFormatRepository serializerFormatRepository;
 
     /**
      * Constructor.
      */
     @Autowired
-    private DataLoaderConfig(final MessageFormatRepository messageFormatRepository, final UserRepository userRepository) {
+    private DataLoaderConfig(
+        final MessageFormatRepository messageFormatRepository,
+        final UserRepository userRepository,
+        final PartitioningStrategyRepository partitioningStrategyRepository,
+        final SerializerFormatRepository serializerFormatRepository) {
         this.messageFormatRepository = messageFormatRepository;
         this.userRepository = userRepository;
+        this.partitioningStrategyRepository = partitioningStrategyRepository;
+        this.serializerFormatRepository = serializerFormatRepository;
     }
 
     /**
@@ -70,6 +85,8 @@ private DataLoaderConfig(final MessageFormatRepository messageFormatRepository,
     private void createData() {
         createDefaultUser();
         createDefaultMessageFormats();
+        createDefaultPartitioningStrategies();
+        createDefaultSerializationFormats();
     }
 
     /**
@@ -119,11 +136,54 @@ private void createDefaultMessageFormats() {
         }
     }
 
+    /**
+     * Creates default partitioning strategies.
+     */
+    private void createDefaultPartitioningStrategies() {
+        final Map<String, String> defaultEntries = new HashMap<>();
+        defaultEntries.put("Default Partitioner", DefaultPartitioner.class.getName());
+
+        // Create if needed.
+        for (final Map.Entry<String, String> entry : defaultEntries.entrySet()) {
+            PartitioningStrategy partitioningStrategy = partitioningStrategyRepository.findByName(entry.getKey());
+            if (partitioningStrategy == null) {
+                partitioningStrategy = new PartitioningStrategy();
+            }
+            partitioningStrategy.setName(entry.getKey());
+            partitioningStrategy.setClasspath(entry.getValue());
+            partitioningStrategy.setJar("n/a");
+            partitioningStrategy.setDefault(true);
+            partitioningStrategyRepository.save(partitioningStrategy);
+        }
+    }
+
+    /**
+     * Creates default serialization formats.
+     */
+    private void createDefaultSerializationFormats() {
+        final Map<String, String> defaultEntries = new HashMap<>();
+        defaultEntries.put("String Serializer", StringTransformer.class.getName());
+        defaultEntries.put("Long Serializer", LongTransformer.class.getName());
+
+        // Create if needed.
+        for (final Map.Entry<String, String> entry : defaultEntries.entrySet()) {
+            SerializerFormat serializerFormat = serializerFormatRepository.findByName(entry.getKey());
+            if (serializerFormat == null) {
+                serializerFormat = new SerializerFormat();
+            }
+            serializerFormat.setName(entry.getKey());
+            serializerFormat.setClasspath(entry.getValue());
+            serializerFormat.setJar("n/a");
+            serializerFormat.setDefault(true);
+            serializerFormatRepository.save(serializerFormat);
+        }
+    }
+
     /**
      * Run on startup.
      */
     @Override
-    public void run(final ApplicationArguments args) throws Exception {
+    public void run(final ApplicationArguments args) {
         createData();
     }
 }

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

@@ -27,7 +27,9 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.hubspot.jackson.datatype.protobuf.ProtobufModule;
+import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.sourcelab.kafka.webview.ui.manager.encryption.SecretManager;
@@ -40,6 +42,7 @@
 import org.sourcelab.kafka.webview.ui.manager.plugin.UploadManager;
 import org.sourcelab.kafka.webview.ui.manager.sasl.SaslUtility;
 import org.sourcelab.kafka.webview.ui.plugin.filter.RecordFilter;
+import org.sourcelab.kafka.webview.ui.plugin.serializer.SerializerTransformer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.jackson.Jackson2ObjectMapperBuilderCustomizer;
 import org.springframework.context.annotation.Bean;
@@ -87,6 +90,39 @@ public PluginFactory<RecordFilter> getRecordFilterPluginFactory(final AppPropert
         return new PluginFactory<>(jarDirectory, RecordFilter.class);
     }
 
+    /**
+     * PluginFactory for creating instances of Partitioners.
+     * @param appProperties Definition of app properties.
+     * @return PluginFactory for Partitioners.
+     */
+    @Bean
+    public PluginFactory<Partitioner> getPartitionerPluginFactory(final AppProperties appProperties) {
+        final String jarDirectory = appProperties.getUploadPath() + "/partitioners";
+        return new PluginFactory<>(jarDirectory, Partitioner.class);
+    }
+
+    /**
+     * PluginFactory for creating instances of Serializers.
+     * @param appProperties Definition of app properties.
+     * @return PluginFactory for Serializers.
+     */
+    @Bean
+    public PluginFactory<Serializer> getSerializerPluginFactory(final AppProperties appProperties) {
+        final String jarDirectory = appProperties.getUploadPath() + "/serializers";
+        return new PluginFactory<>(jarDirectory, Serializer.class);
+    }
+
+    /**
+     * PluginFactory for creating instances of Serializer Transformers.
+     * @param appProperties Definition of app properties.
+     * @return PluginFactory for Serializer Transformers.
+     */
+    @Bean
+    public PluginFactory<SerializerTransformer> getSerializerTransformerPluginFactory(final AppProperties appProperties) {
+        final String jarDirectory = appProperties.getUploadPath() + "/serializers";
+        return new PluginFactory<>(jarDirectory, SerializerTransformer.class);
+    }
+
     /**
      * For handling secrets, symmetrical encryption.
      * @param appProperties Definition of app properties.