Fix sideline zk configuration, docs, zulu support & flapper (#122)

* Fix root config key, it shouldn't be plural * Fix doc blocks for config * Drop using javax because it's not in zulu by default * Fix flapping test due to statics in MockConsumer * Add other jdks to travis build matrix

Author: stanlemon

.travis.yml

@@ -1,6 +1,8 @@
 language: java
 jdk:
  - oraclejdk8
+ - oraclejdk11
+ - openjdk11
 sudo: false
 cache:
  directories:

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

@@ -567,19 +567,35 @@ private void attemptToComplete() {
  // Get current state and compare it against our ending state
  final ConsumerState currentState = consumer.getCurrentState();
 
+ if (currentState == null) {
+ logger.error("Attempting to flush state when the current state is null.");
+ return;
+ }
+
  // Compare it against our ending state
  for (final ConsumerPartition consumerPartition : currentState.getConsumerPartitions()) {
  // currentOffset contains the last "committed" offset our consumer has fully processed
- final long currentOffset = currentState.getOffsetForNamespaceAndPartition(consumerPartition);
+ final Long currentOffset = currentState.getOffsetForNamespaceAndPartition(consumerPartition);
 
  // endingOffset contains the last offset we want to process.
- final long endingOffset = endingState.getOffsetForNamespaceAndPartition(consumerPartition);
+ final Long endingOffset = endingState.getOffsetForNamespaceAndPartition(consumerPartition);
+
+ if (currentOffset == null) {
+ logger.error("Current offset is null");
+ return;
+ }
+
+ if (endingOffset == null) {
+ logger.error("Ending offset is null");
+ return;
+ }
 
  // If the current offset is < ending offset
  if (currentOffset < endingOffset) {
  // Then we cannot end
  return;
  }
+
  // Log that this partition is finished, and make sure we unsubscribe from it.
  if (consumer.unsubscribeConsumerPartition(consumerPartition)) {
  logger.debug(

src/main/java/com/salesforce/storm/spout/dynamic/kafka/Consumer.java

@@ -455,8 +455,12 @@ public ConsumerState flushConsumerState() {
  // Get the current state
  final ConsumerState consumerState = partitionOffsetsManager.getCurrentState();
 
- // Persist each partition offset
+ if (consumerState == null) {
+ logger.error("Attempting to flush state when the current state is null.");
+ return null;
+ }
 
+ // Persist each partition offset
  for (Map.Entry<ConsumerPartition, Long> entry : consumerState.entrySet()) {
  final ConsumerPartition consumerPartition = entry.getKey();
  final long lastFinishedOffset = entry.getValue();

src/main/java/com/salesforce/storm/spout/sideline/config/SidelineConfig.java

@@ -39,13 +39,13 @@
 public class SidelineConfig {
 
  /**
- * (List|String) Defines one or more sideline trigger(s) (if any) to use.
+ * (String) Defines a sideline trigger (if any) to use.
  * Should be a fully qualified class path that implements thee SidelineTrigger interface.
  */
  @ConfigDocumentation(
  category = ConfigDocumentation.Category.SIDELINE,
- description = "Defines one or more sideline trigger(s) (if any) to use. "
- + "Should be a fully qualified class path that implements thee SidelineTrigger interface.",
+ description = "Defines a sideline trigger (if any) to use. "
+ + "Should be a fully qualified class path that implements the SidelineTrigger interface.",
  type = String.class
  )
  public static final String TRIGGER_CLASS = "sideline.trigger_class";

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

@@ -71,7 +71,7 @@ public class Config {
  + "Example: \"/sideline-trigger/my-topology/my-topic\"",
  type = String.class
  )
- public static final String ZK_ROOT = PREFIX + "roots";
+ public static final String ZK_ROOT = PREFIX + "root";
 
  /**
  * (Integer) Zookeeper session timeout.

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

@@ -47,10 +47,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.xml.bind.DatatypeConverter;
 import java.io.IOException;
-import java.io.UnsupportedEncodingException;
+import java.math.BigInteger;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.time.LocalDateTime;
@@ -293,7 +293,7 @@ private SidelineRequest buildSidelineRequest(final TriggerEvent triggerEvent) {
  logger.info("Creating a sideline request with id {} and step {}", sidelineRequest.id, sidelineRequest.step);
 
  return sidelineRequest;
- } catch (NoSuchAlgorithmException | UnsupportedEncodingException ex) {
+ } catch (NoSuchAlgorithmException ex) {
  logger.error("Unable to generate an identifier for this request, cowardly refusing to proceed! {}", ex);
  return null;
  }
@@ -309,12 +309,10 @@ private SidelineRequest buildSidelineRequest(final TriggerEvent triggerEvent) {
  private SidelineRequestIdentifier generateSidelineRequestIdentifier(
  final TriggerEvent triggerEvent,
  final FilterChainStep step
- ) throws NoSuchAlgorithmException, UnsupportedEncodingException {
+ ) throws NoSuchAlgorithmException {
  final String json = gson.toJson(step);
 
- final StringBuilder identifier = new StringBuilder(
- DatatypeConverter.printHexBinary(MessageDigest.getInstance("MD5").digest(json.getBytes("UTF-8")))
- );
+ final StringBuilder identifier = new StringBuilder(generateIdFromJson(json));
 
  // If we were provided a date time in the event, append the time stamp of that event to the identifier
  if (triggerEvent.getCreatedAt() != null) {
@@ -327,6 +325,14 @@ private SidelineRequestIdentifier generateSidelineRequestIdentifier(
  return new SidelineRequestIdentifier(identifier.toString());
  }
 
+ private String generateIdFromJson(final String dataJson) throws NoSuchAlgorithmException {
+ // Use the data map, which should be things unique to define this criteria to generate our id
+ final MessageDigest md5 = MessageDigest.getInstance("MD5");
+ md5.update(StandardCharsets.UTF_8.encode(dataJson));
+ final String id = String.format("%032x", new BigInteger(1, md5.digest()));
+ return id;
+ }
+
  /**
  * Create a trigger event from the provided data.
  * @return Trigger event

src/test/java/com/salesforce/storm/spout/dynamic/VirtualSpoutTest.java

@@ -49,6 +49,7 @@
 import org.apache.storm.task.TopologyContext;
 import org.apache.storm.tuple.Values;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.lang.reflect.Field;
@@ -79,6 +80,14 @@
  */
 public class VirtualSpoutTest {
 
+ /**
+ * Before each test reset the static on the MockConsumer.
+ */
+ @BeforeEach
+ void resetMockConsumerPartitions() {
+ MockConsumer.partitions = Arrays.asList(1);
+ }
+
  /**
  * Verify that constructor args get set appropriately.
  */