Remove FilterChainBuilder and serialize the steps directly (#130)

* Remove FilterChainBuilder and serialize the steps directly * Simplify and cleanup serialization * Switch to checkNotNull * Move TriggerEvent identification generation into it's self

Author: stanlemon

pom.xml

@@ -374,7 +374,7 @@
  <plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-checkstyle-plugin</artifactId>
- <version>2.17</version>
+ <version>3.1.0</version>
  <executions>
  <execution>
  <id>checkstyle-validate</id>
@@ -467,12 +467,12 @@
  <plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-checkstyle-plugin</artifactId>
- <version>2.17</version>
+ <version>3.1.0</version>
  <dependencies>
  <dependency>
  <groupId>com.puppycrawl.tools</groupId>
  <artifactId>checkstyle</artifactId>
- <version>8.18</version>
+ <version>8.26</version>
  </dependency>
  </dependencies>
  </plugin>

script/checkstyle-ruleset.xml

@@ -20,6 +20,11 @@
 
  <module name="SuppressWarningsFilter" />
 
+ <module name="LineLength">
+ <property name="max" value="140"/>
+ <property name="ignorePattern" value="^package.*|^import.*|a href|href|http://|https://|ftp://"/>
+ </module>
+
  <module name="TreeWalker">
  <module name="SuppressWarningsHolder" />
  <module name="OuterTypeFilename"/>
@@ -33,10 +38,6 @@
  <property name="allowByTailComment" value="true"/>
  <property name="allowNonPrintableEscapes" value="true"/>
  </module>
- <module name="LineLength">
- <property name="max" value="140"/>
- <property name="ignorePattern" value="^package.*|^import.*|a href|href|http://|https://|ftp://"/>
- </module>
  <module name="AvoidStarImport"/>
  <module name="RedundantImport"/>
  <module name="UnusedImports"/>
@@ -175,7 +176,6 @@
  <property name="allowMissingParamTags" value="true"/>
  <property name="allowMissingThrowsTags" value="true"/>
  <property name="allowMissingReturnTag" value="true"/>
- <property name="minLineCount" value="2"/>
  <property name="allowedAnnotations" value="Override, Test"/>
  <property name="allowThrowsTagsForSubclasses" value="true"/>
  </module>

src/main/java/com/salesforce/storm/spout/documentation/ConfigDocumentation.java

@@ -89,5 +89,6 @@ public String toString() {
  /**
  * Default class type for use on the type field.
  */
- final class Default {}
+ final class Default {
+ }
 }

src/main/java/com/salesforce/storm/spout/dynamic/JSON.java

@@ -30,6 +30,7 @@
 import com.salesforce.storm.spout.dynamic.filter.FilterChainStep;
 import com.salesforce.storm.spout.sideline.persistence.FilterChainStepSerializer;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -47,15 +48,26 @@ public class JSON {
  */
  private final Gson gson;
 
+ /**
+ * Create JSON serializer/deserializer instance.
+ */
+ public JSON() {
+ this(new HashMap<>());
+ }
+
  /**
  * Create JSON serializer/deserializer instance with default configuration.
  * @param config configuration.
  */
  public JSON(final Map<String, Object> config) {
- this.gson = new GsonBuilder()
- .setDateFormat("yyyy-MM-dd HH:mm:ss")
- .registerTypeAdapter(FilterChainStep.class, new FilterChainStepSerializer(config))
- .create();
+ final GsonBuilder gsonBuilder = new GsonBuilder()
+ .setDateFormat("yyyy-MM-dd HH:mm:ss");
+
+ if (!config.isEmpty()) {
+ gsonBuilder.registerTypeAdapter(FilterChainStep.class, new FilterChainStepSerializer(config));
+ }
+
+ this.gson = gsonBuilder.create();
  }
 
  /**

src/main/java/com/salesforce/storm/spout/sideline/recipes/trigger/FilterChainStepBuilder.java renamed to src/main/java/com/salesforce/storm/spout/sideline/recipes/trigger/KeyFilter.java

@@ -25,19 +25,26 @@
 
 package com.salesforce.storm.spout.sideline.recipes.trigger;
 
+import com.salesforce.storm.spout.dynamic.Message;
 import com.salesforce.storm.spout.dynamic.filter.FilterChainStep;
 
-import java.util.Map;
+import java.util.List;
 
 /**
- * Given a map of a data from a {@link TriggerEvent} implementations of this generate {@link FilterChainStep} instances.
+ * Filter for filtering messages by their key.
  */
-public interface FilterChainStepBuilder {
+public class KeyFilter implements FilterChainStep {
 
- /**
- * Given a map of a data from a {@link TriggerEvent} implementations of this generate {@link FilterChainStep} instances.
- * @param data data from a trigger event.
- * @return filter chain step.
- */
- FilterChainStep build(final Map<String,Object> data);
+ private final List<String> filteredKeys;
+
+ public KeyFilter(final List<String> filteredKeys) {
+ this.filteredKeys = filteredKeys;
+ }
+
+ @Override
+ public boolean filter(Message message) {
+ final String key = String.valueOf(message.getValues().get(0));
+ // If the key for this message is contained in our list we filter it out
+ return filteredKeys.contains(key);
+ }
 }

src/main/java/com/salesforce/storm/spout/sideline/recipes/trigger/TriggerEvent.java

@@ -26,10 +26,13 @@
 package com.salesforce.storm.spout.sideline.recipes.trigger;
 
 import com.google.common.base.Preconditions;
+import com.salesforce.storm.spout.dynamic.JSON;
+import com.salesforce.storm.spout.dynamic.Tools;
+import com.salesforce.storm.spout.dynamic.filter.FilterChainStep;
 import com.salesforce.storm.spout.sideline.trigger.SidelineType;
 
 import java.time.LocalDateTime;
-import java.util.Map;
+import java.time.ZoneOffset;
 import java.util.Objects;
 
 /**
@@ -40,7 +43,7 @@ public class TriggerEvent {
 
  private SidelineType type;
 
- private Map<String,Object> data;
+ private FilterChainStep filterChainStep;
 
  private LocalDateTime createdAt;
 
@@ -55,49 +58,80 @@ public class TriggerEvent {
  /**
  * An event to a {@link com.salesforce.storm.spout.sideline.trigger.SidelineTrigger} that communicates the desired type of
  * sideline state for the system.
- *
+ * <p>
  * When you create a TriggerEvent in Zookeeper always set processed = false, the trigger implementation will flip this to true
  * after it has been picked up and handled by the trigger. This allows you to distinguish an event that's been handled by the
  * trigger and one that has not.
  *
- * @param type sideline type.
- * @param data data bag of key value pairs.
- * @param createdAt when the event was created.
- * @param createdBy who created the event.
+ * @param type  sideline type.
+ * @param filterChainStep filter chain step
+ * @param createdAt  when the event was created.
+ * @param createdBy  who created the event.
  * @param description a description of the reason for the sideline request.
- * @param processed whether or not the event (in its current state has been processed)
- * @param updatedAt Timestamp the event was last updated.
+ * @param processed  whether or not the event (in its current state has been processed)
+ * @param updatedAt  Timestamp the event was last updated.
  */
  public TriggerEvent(
  final SidelineType type,
- final Map<String,Object> data,
+ final FilterChainStep filterChainStep,
  final LocalDateTime createdAt,
  final String createdBy,
  final String description,
  final boolean processed,
  final LocalDateTime updatedAt
  ) {
  Preconditions.checkNotNull(type, "Type is required.");
- Preconditions.checkNotNull(data, "Data payload is required (But we do accept empty maps!).");
+ Preconditions.checkNotNull(filterChainStep, "FilterChainStep is required.");
  Preconditions.checkNotNull(createdAt, "Created at time is required.");
  Preconditions.checkNotNull(createdBy, "Created by is required.");
  Preconditions.checkNotNull(description, "Description is required.");
 
  this.type = type;
- this.data = data;
+ this.filterChainStep = filterChainStep;
  this.createdAt = createdAt;
  this.createdBy = createdBy;
  this.description = description;
  this.processed = processed;
  this.updatedAt = updatedAt;
  }
 
+ /**
+ * Generate an identifier for this {@link TriggerEvent}.
+ *
+ * The identifier generated here is a JSON serialized version of the {@link FilterChainStep} which we then make
+ * an md5 hash of. If we have a created date we tack on it's millis to the end for extra uniqueness. It's the
+ * {@link FilterChainStep} that is the important part of a {@link TriggerEvent} and the assumption here is that
+ * there are properties that make the given filter unique. For example, if you're filtering out a tenant on a
+ * multi tenant stream you probably have a generic filter which has a property to hold the given tenant.
+ *
+ * If you don't have uniqueness represented in the properties of your filter this whole recipe probably isn't a
+ * great fit for your use case.
+ *
+ * @return string representing the event.
+ */
+ public String getIdentifier() {
+ // Note: This only works for to() because we only need the implementation name for from()
+ final String data = (new JSON()).to(getFilterChainStep());
+
+ final StringBuilder identifier = new StringBuilder(Tools.makeMd5Hash(data));
+
+ // If we were provided a date time in the event, append the time stamp of that event to the identifier
+ if (this.getCreatedAt() != null) {
+ identifier.append("-");
+ identifier.append(
+ this.getCreatedAt().atZone(ZoneOffset.UTC).toInstant().toEpochMilli()
+ );
+ }
+
+ return identifier.toString();
+ }
+
  public SidelineType getType() {
  return type;
  }
 
- public Map<String,Object> getData() {
- return data;
+ public FilterChainStep getFilterChainStep() {
+ return filterChainStep;
  }
 
  public LocalDateTime getCreatedAt() {
@@ -131,7 +165,7 @@ public boolean equals(Object obj) {
  TriggerEvent that = (TriggerEvent) obj;
  return processed == that.processed
  && type == that.type
- && Objects.equals(data, that.data)
+ && Objects.equals(filterChainStep, that.filterChainStep)
  && Objects.equals(createdAt, that.createdAt)
  && Objects.equals(createdBy, that.createdBy)
  && Objects.equals(description, that.description)
@@ -140,14 +174,14 @@ public boolean equals(Object obj) {
 
  @Override
  public int hashCode() {
- return Objects.hash(type, data, createdAt, createdBy, description);
+ return Objects.hash(type, filterChainStep, createdAt, createdBy, description);
  }
 
  @Override
  public String toString() {
  return "TriggerEvent{"
  + "type=" + type
- + ", data=" + data
+ + ", filterChainStep=" + filterChainStep
  + ", createdAt=" + createdAt
  + ", createdBy='" + createdBy + '\''
  + ", description='" + description + '\''

src/main/java/com/salesforce/storm/spout/sideline/recipes/trigger/TriggerEventHelper.java

@@ -28,6 +28,7 @@
 import com.google.common.base.Preconditions;
 import com.salesforce.storm.spout.dynamic.JSON;
 import com.salesforce.storm.spout.dynamic.Tools;
+import com.salesforce.storm.spout.dynamic.filter.FilterChainStep;
 import com.salesforce.storm.spout.dynamic.persistence.zookeeper.CuratorFactory;
 import com.salesforce.storm.spout.dynamic.persistence.zookeeper.CuratorHelper;
 import com.salesforce.storm.spout.sideline.recipes.trigger.zookeeper.Config;
@@ -83,34 +84,34 @@ public TriggerEventHelper(final Map<String, Object> config) {
  /**
  * Create a {@link TriggerEvent} to start a sideline.
  *
- * @param data data for the trigger event, this will be handed off to the {@link FilterChainStepBuilder}.
+ * @param filterChainStep a {@link FilterChainStep} object to be applied for the trigger event.
  * @param createdBy the name or a way of identifying who or what started the sideline.
  * @param reason the reason for starting the sideline.
  * @return the identifier of the sideline.
  */
  public String startTriggerEvent(
- final Map<String, Object> data,
+ final FilterChainStep filterChainStep,
  final String createdBy,
  final String reason
  ) {
- Preconditions.checkArgument(
- data != null && !data.isEmpty(),
- "TriggerEvent's require data"
+ Preconditions.checkNotNull(
+ filterChainStep,
+ "TriggerEvent's require a FilterChainStep"
  );
 
  final LocalDateTime createdAt = LocalDateTime.now();
 
  final TriggerEvent triggerEvent = new TriggerEvent(
  SidelineType.START,
- data,
+ filterChainStep,
  createdAt,
  createdBy,
  reason,
  false,
  createdAt
  );
 
- final String id = getMd5Hash(data);
+ final String id = getMd5Hash(filterChainStep);
  final String path = getZkRoot() + "/" + id;
 
  logger.info("Sending trigger event to start sideline {}", id);
@@ -160,7 +161,7 @@ private void updateTriggerEventType(final String id, SidelineType type) {
 
  final TriggerEvent triggerEvent = new TriggerEvent(
  type,
- originalTriggerEvent.getData(),
+ originalTriggerEvent.getFilterChainStep(),
  originalTriggerEvent.getCreatedAt(),
  originalTriggerEvent.getCreatedBy(),
  originalTriggerEvent.getDescription(),
@@ -173,8 +174,8 @@ private void updateTriggerEventType(final String id, SidelineType type) {
  logger.info("Saved to {}", path);
  }
 
- private String getMd5Hash(final Map<String, Object> data) {
- return Tools.makeMd5Hash(new JSON(new HashMap<>()).to(data));
+ private String getMd5Hash(final FilterChainStep filterChainStep) {
+ return Tools.makeMd5Hash(new JSON(new HashMap<>()).to(filterChainStep));
  }
 
  private String getZkRoot() {

src/main/java/com/salesforce/storm/spout/sideline/recipes/trigger/zookeeper/Config.java

@@ -26,7 +26,6 @@
 package com.salesforce.storm.spout.sideline.recipes.trigger.zookeeper;
 
 import com.salesforce.storm.spout.documentation.ConfigDocumentation;
-import com.salesforce.storm.spout.sideline.recipes.trigger.FilterChainStepBuilder;
 
 import java.util.List;
 
@@ -40,15 +39,6 @@ public class Config {
  */
  public static final String PREFIX = "sideline.zookeeper_watch_trigger.";
 
- /**
- * (String) Class name for the class of the {@link FilterChainStepBuilder} instance.
- */
- @ConfigDocumentation(
- description = "Class name for the class of the FilterChainStepBuilder instance.",
- type = String.class
- )
- public static final String FILTER_CHAIN_STEP_BUILDER_CLASS = PREFIX + "filter_chain_step_builder_class";
-
  /**
  * (List[String]) Holds a list of Zookeeper server Hostnames + Ports in the following format:
  * ["zkhost1:2181", "zkhost2:2181", ...]