diff --git a/pom.xml b/pom.xml index 5b9f902..4a846d8 100644 --- a/pom.xml +++ b/pom.xml @@ -8,17 +8,24 @@ kafka-webview-examples 1.0.0 + + 4.1.0 + http://packages.confluent.io/maven/ + + + + + confluent + Confluent + ${confluent.maven.repo} + + + + - - org.sourcelab - kafka-webview-plugin - 1.0.0 - provided - - @@ -27,6 +34,12 @@ 2.0.0 provided + + + io.confluent + kafka-avro-serializer + ${avro-registry-version} + @@ -76,4 +89,4 @@ - + \ No newline at end of file diff --git a/src/main/java/examples/deserializer/ExampleDeserializer.java b/src/main/java/examples/deserializer/ExampleDeserializer.java deleted file mode 100644 index 86419e0..0000000 --- a/src/main/java/examples/deserializer/ExampleDeserializer.java +++ /dev/null @@ -1,32 +0,0 @@ -package examples.deserializer; - -import org.apache.kafka.common.serialization.Deserializer; - -import java.nio.charset.StandardCharsets; -import java.util.Map; - -/** - * Dummy implementation that has little to no real world value. Simply prefixes any stored - * string values with a prefixed string. - */ -public class ExampleDeserializer implements Deserializer { - - @Override - public void configure(final Map configs, final boolean isKey) { - // Not used in this implementation. - } - - @Override - public String deserialize(final String topic, final byte[] data) { - // Convert to string - final String stringVal = new String(data, StandardCharsets.UTF_8); - - // Prefix it - return "Prefixed Value: " + stringVal; - } - - @Override - public void close() { - // Not used in this implementation. - } -} diff --git a/src/main/java/examples/filter/LowOffsetFilter.java b/src/main/java/examples/filter/LowOffsetFilter.java deleted file mode 100644 index 861ff03..0000000 --- a/src/main/java/examples/filter/LowOffsetFilter.java +++ /dev/null @@ -1,49 +0,0 @@ -package examples.filter; - -import org.sourcelab.kafka.webview.ui.plugin.filter.RecordFilter; - -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -/** - * Example filter that removes low offsets. - */ -public class LowOffsetFilter implements RecordFilter { - - private static final String OPTION_MIN_OFFSET = "Minimum Offset"; - - /** - * Defaults to 600. - */ - private Long minimumOffset = 600L; - - @Override - public Set getOptionNames() { - final Set options = new HashSet<>(); - options.add(OPTION_MIN_OFFSET); - return options; - } - - /** - * Configure this class. - * @param consumerConfigs Consumer configuration in key/value pairs - * @param filterOptions User defined filter options. - */ - @Override - public void configure(final Map consumerConfigs, final Map filterOptions) { - if (filterOptions.containsKey(OPTION_MIN_OFFSET)) { - minimumOffset = Long.valueOf(filterOptions.get(OPTION_MIN_OFFSET)); - } - } - - @Override - public boolean includeRecord(final String topic, final int partition, final long offset, final Object key, final Object value) { - return offset > minimumOffset; - } - - @Override - public void close() { - // Not used. - } -} diff --git a/src/main/java/examples/filter/ModulusOffsetFilter.java b/src/main/java/examples/filter/ModulusOffsetFilter.java deleted file mode 100644 index 3efe897..0000000 --- a/src/main/java/examples/filter/ModulusOffsetFilter.java +++ /dev/null @@ -1,64 +0,0 @@ -package examples.filter; - -import org.sourcelab.kafka.webview.ui.plugin.filter.RecordFilter; - -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -/** - * Example filter that filters messages by looking at the offset. It can take in an optional - * filterOptions configuration value and use it to calculate the modulus of the offset. - */ -public class ModulusOffsetFilter implements RecordFilter { - private static final String OPTION_MODULUS = "modulus"; - - /** - * By default filter even. - */ - private Integer modulus = 2; - - @Override - public Set getOptionNames() { - final Set options = new HashSet<>(); - options.add(OPTION_MODULUS); - return options; - } - - /** - * Configure this class. - * @param consumerConfigs Consumer configuration in key/value pairs - * @param filterOptions User defined filter options. - */ - @Override - public void configure(final Map consumerConfigs, final Map filterOptions) { - if (filterOptions.containsKey(OPTION_MODULUS)) { - modulus = Integer.valueOf(filterOptions.get(OPTION_MODULUS)); - } - } - - /** - * Define the filter behavior. - * A return value of TRUE means the record WILL be shown. - * A return value of FALSE means the record will NOT be shown. - * - * @param topic Name of topic the message came from. - * @param partition Partition the message came from. - * @param offset Offset the message came from. - * @param key Deserialized Key object. - * @param value Deserialized Value object. - * @return True means the record WILL be shown. False means the record will NOT be shown. - */ - @Override - public boolean includeRecord(final String topic, final int partition, final long offset, final Object key, final Object value) { - return offset % modulus == 0; - } - - /** - * Called on closing. - */ - @Override - public void close() { - // Nothing to do here! - } -} diff --git a/src/main/java/examples/filter/StringSearchFilter.java b/src/main/java/examples/filter/StringSearchFilter.java deleted file mode 100644 index 00ed0bc..0000000 --- a/src/main/java/examples/filter/StringSearchFilter.java +++ /dev/null @@ -1,91 +0,0 @@ -package examples.filter; - -import org.sourcelab.kafka.webview.ui.plugin.filter.RecordFilter; - -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -/** - * Example filter that only passes records that contain a configured String value. - */ -public class StringSearchFilter implements RecordFilter { - /** - * Defines the name of our configurable option to define the filter value. - */ - private final String OPTION_FILTER_VALUE = "FilterValue"; - - /** - * The configured filter value str. - */ - private String filterValueStr = null; - - /** - * Define names of configurable options. - * These names will be passed up to the User Interface and allow the user to define them. - * When configure() is called below, these same names will be returned, along with the user defined values, - * in the filterOptions argument. - * - * Since the UI provides no validation on these user defined values, best practices dictate that your implementation - * should gracefully handle when these are not set. - * - * @return Set of option names. - */ - @Override - public Set getOptionNames() { - final Set optionNames = new HashSet<>(); - optionNames.add(OPTION_FILTER_VALUE); - return optionNames; - } - - /** - * Configure this class. - * @param consumerConfigs Consumer configuration in key/value pairs - * @param filterOptions User defined filter options. - */ - @Override - public void configure(final Map consumerConfigs, final Map filterOptions) { - if (filterOptions.containsKey(OPTION_FILTER_VALUE)) { - filterValueStr = filterOptions.get(OPTION_FILTER_VALUE); - } else { - throw new RuntimeException("Unconfigured Filter Value!"); - } - } - - /** - * Define the filter behavior. - * A return value of TRUE means the record WILL be shown. - * A return value of FALSE means the record will NOT be shown. - * - * @param topic Name of topic the message came from. - * @param partition Partition the message came from. - * @param offset Offset the message came from. - * @param key Deserialized Key object. - * @param value Deserialized Value object. - * @return True means the record WILL be shown. False means the record will NOT be shown. - */ - @Override - public boolean includeRecord(final String topic, final int partition, final long offset, final Object key, final Object value) { - // Sanity check. This filter only works on String values. - if (value == null || !(value instanceof String)) { - // Fail safe, pass if it's not of type String. - return true; - } - - // If the value contains our string - if (((String) value).contains(filterValueStr)) { - // we should show it. - return true; - } - // Otherwise filter it. - return false; - } - - /** - * Called on closing. - */ - @Override - public void close() { - // No cleanup required. - } -}