Merge pull request #6 from SourceLabOrg/sp/jedisImplementation

Add implementation using Jedis client library

Author: Crim

CHANGELOG.md

@@ -2,5 +2,11 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## 1.1.0 (07/24/2020)
+- Add Jedis implementation. Spout defaults to using the Lettuce redis library, but you can configure
+ to use the Jedis library instead via the `RedisStreamSpoutConfig.withJedisClientLibrary()` method.
+- Bugfix on Spout deploy, consumer thread started during `open()` lifecycle call instead of `activate()`. 
+- Bugfix on Spout restart, resume consuming first from consumers personal pending list.
+
 ## 1.0.0 (07/20/2020)
 - Initial release!

pom.xml

@@ -6,7 +6,7 @@
 
  <groupId>org.sourcelab.storm.spout</groupId>
  <artifactId>redis-stream-spout</artifactId>
- <version>1.0.0</version>
+ <version>1.1.0</version>
 
  <!-- Module Description and Ownership -->
  <name>Redis Streams Spout for Apache Storm.</name>
@@ -48,6 +48,7 @@
 
  <!-- Redis Client -->
  <lettuceVersion>5.3.1.RELEASE</lettuceVersion>
+ <jedisVersion>3.2.0</jedisVersion>
 
  <!-- Define which version of JUnit 5 to -->
  <junit5Version>5.6.2</junit5Version>
@@ -78,6 +79,11 @@
  <artifactId>lettuce-core</artifactId>
  <version>${lettuceVersion}</version>
  </dependency>
+ <dependency>
+ <groupId>redis.clients</groupId>
+ <artifactId>jedis</artifactId>
+ <version>${jedisVersion}</version>
+ </dependency>
 
  <!-- Testing Dependencies -->
  <dependency>
@@ -120,6 +126,14 @@
  <version>${testContainersVersion}</version>
  <scope>test</scope>
  </dependency>
+
+ <!-- Async Testing Helper -->
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>4.0.3</version>
+ <scope>test</scope>
+ </dependency>
  </dependencies>
 
  <build>

src/main/java/org/sourcelab/storm/spout/redis/RedisStreamSpout.java

@@ -8,8 +8,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.sourcelab.storm.spout.redis.client.Client;
+import org.sourcelab.storm.spout.redis.client.ClientFactory;
 import org.sourcelab.storm.spout.redis.client.Consumer;
-import org.sourcelab.storm.spout.redis.client.LettuceClient;
+import org.sourcelab.storm.spout.redis.client.lettuce.LettuceClient;
 import org.sourcelab.storm.spout.redis.funnel.ConsumerFunnel;
 import org.sourcelab.storm.spout.redis.funnel.MemoryFunnel;
 import org.sourcelab.storm.spout.redis.funnel.SpoutFunnel;
@@ -83,16 +84,8 @@ public void open(
  // Create funnel instance.
  this.funnel = new MemoryFunnel(config, spoutConfig, topologyContext);
 
- // Create consumer and client
- final int taskIndex = topologyContext.getThisTaskIndex();
- final Client client = new LettuceClient(config, taskIndex);
- final Consumer consumer = new Consumer(config, client, (ConsumerFunnel) funnel);
-
- // Create background consuming thread.
- consumerThread = new Thread(
- consumer,
- "RedisStreamSpout-ConsumerThread[" + taskIndex + "]"
- );
+ // Create and start consumer thread.
+ createAndStartConsumerThread();
  }
 
  @Override
@@ -103,12 +96,15 @@ public void close() {
 
  @Override
  public void activate() {
- if (consumerThread.isAlive()) {
+ // If the thread is already running and alive
+ if (consumerThread != null && consumerThread.isAlive()) {
  // No-op. It's already running, and deactivate() is a no-op for us.
  return;
  }
- // Start thread, this should return immediately, but start a background processing thread.
- consumerThread.start();
+
+ // If we haven't created the consumer thread yet, or it has previously died.
+ // Create and start it
+ createAndStartConsumerThread();
  }
 
  @Override
@@ -186,4 +182,21 @@ public void declareOutputFields(final OutputFieldsDeclarer declarer) {
  public Map<String, Object> getComponentConfiguration() {
  return new HashMap<>();
  }
+
+ /**
+ * Create background consumer thread.
+ */
+ private void createAndStartConsumerThread() {
+ // Create consumer and client
+ final int taskIndex = topologyContext.getThisTaskIndex();
+ final Client client = new ClientFactory().createClient(config, taskIndex);
+ final Consumer consumer = new Consumer(config, client, (ConsumerFunnel) funnel);
+
+ // Create background consuming thread.
+ consumerThread = new Thread(
+ consumer,
+ "RedisStreamSpout-ConsumerThread[" + taskIndex + "]"
+ );
+ consumerThread.start();
+ }
 }

src/main/java/org/sourcelab/storm/spout/redis/RedisStreamSpoutConfig.java

@@ -1,5 +1,6 @@
 package org.sourcelab.storm.spout.redis;
 
+import org.sourcelab.storm.spout.redis.client.ClientType;
 import org.sourcelab.storm.spout.redis.failhandler.NoRetryHandler;
 
 import java.io.Serializable;
@@ -70,6 +71,11 @@ public class RedisStreamSpoutConfig implements Serializable {
  */
  private final boolean metricsEnabled;
 
+ /**
+ * Defines which underlying client library/implementation to use.
+ */
+ private final ClientType clientType;
+
  /**
  * Constructor.
  * Use Builder instance.
@@ -85,7 +91,7 @@ private RedisStreamSpoutConfig(
 
  // Other settings
  final int maxConsumePerRead, final int maxTupleQueueSize, final int maxAckQueueSize, final long consumerDelayMillis,
- final boolean metricsEnabled
+ final boolean metricsEnabled, final ClientType clientType
  ) {
  // Connection
  if (redisCluster != null && redisServer != null) {
@@ -117,6 +123,9 @@ private RedisStreamSpoutConfig(
  this.maxAckQueueSize = maxAckQueueSize;
  this.consumerDelayMillis = consumerDelayMillis;
  this.metricsEnabled = metricsEnabled;
+
+ // Client type implementation
+ this.clientType = Objects.requireNonNull(clientType);
  }
 
  public String getStreamKey() {
@@ -150,6 +159,17 @@ public String getConnectString() {
  return redisCluster.getConnectString();
  }
 
+ /**
+ * The URI for connecting to this Redis Server instance with the password masked.
+ * @return URI for the server.
+ */
+ public String getConnectStringMasked() {
+ if (!isConnectingToCluster()) {
+ return redisServer.getConnectStringMasked();
+ }
+ return redisCluster.getConnectStringMasked();
+ }
+
  public int getMaxTupleQueueSize() {
  return maxTupleQueueSize;
  }
@@ -174,6 +194,10 @@ public boolean isMetricsEnabled() {
  return metricsEnabled;
  }
 
+ public ClientType getClientType() {
+ return clientType;
+ }
+
  /**
  * Create a new Builder instance.
  * @return Builder for Configuration instance.
@@ -218,6 +242,12 @@ public static final class Builder {
  private long consumerDelayMillis = 1000L;
  private boolean metricsEnabled = true;
 
+ /**
+ * Underlying library to use.
+ * Defaults to using Lettuce.
+ */
+ private ClientType clientType = ClientType.LETTUCE;
+
  private Builder() {
  }
 
@@ -385,6 +415,27 @@ public Builder withMetricsEnabled(final boolean enabled) {
  return this;
  }
 
+ /**
+ * Configure the spout to use the Lettuce client library for communicating with redis.
+ * @return Builder instance.
+ */
+ public Builder withLettuceClientLibrary() {
+ return withClientType(ClientType.LETTUCE);
+ }
+
+ /**
+ * Configure the spout to use the Jedis client library for communicating with redis.
+ * @return Builder instance.
+ */
+ public Builder withJedisClientLibrary() {
+ return withClientType(ClientType.JEDIS);
+ }
+
+ public Builder withClientType(final ClientType clientType) {
+ this.clientType = Objects.requireNonNull(clientType);
+ return this;
+ }
+
  /**
  * Creates new Configuration instance.
  * @return Configuration instance.
@@ -405,7 +456,10 @@ public RedisStreamSpoutConfig build() {
  tupleConverter, failureHandler,
  // Other settings
  maxConsumePerRead, maxTupleQueueSize, maxAckQueueSize, consumerDelayMillis,
- metricsEnabled
+ metricsEnabled,
+
+ // Underlying client type
+ clientType
  );
  }
  }
@@ -445,6 +499,16 @@ public String getConnectString() {
  .map(RedisServer::getConnectString)
  .collect(Collectors.joining(","));
  }
+
+ /**
+ * The URI for connecting to this Redis Server instance with the password masked.
+ * @return URI for the server.
+ */
+ public String getConnectStringMasked() {
+ return getServers().stream()
+ .map(RedisServer::getConnectStringMasked)
+ .collect(Collectors.joining(","));
+ }
  }
 
  /**
@@ -503,6 +567,21 @@ public String getConnectString() {
  return connectStr;
  }
 
+ /**
+ * The URI for connecting to this Redis Server instance with the password masked.
+ * @return URI for the server.
+ */
+ public String getConnectStringMasked() {
+ String connectStr = "redis://";
+
+ if (getPassword() != null && !getPassword().trim().isEmpty()) {
+ connectStr += "XXXXXX@";
+ }
+ connectStr += getHost() + ":" + getPort();
+
+ return connectStr;
+ }
+
  @Override
  public String toString() {
  return "RedisServer{"

src/main/java/org/sourcelab/storm/spout/redis/client/ClientFactory.java

@@ -0,0 +1,37 @@
+package org.sourcelab.storm.spout.redis.client;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.sourcelab.storm.spout.redis.RedisStreamSpoutConfig;
+import org.sourcelab.storm.spout.redis.client.jedis.JedisClient;
+import org.sourcelab.storm.spout.redis.client.lettuce.LettuceClient;
+
+import java.util.Objects;
+
+/**
+ * Factory for creating the appropriate Client instance based on config.
+ */
+public class ClientFactory {
+ private static final Logger logger = LoggerFactory.getLogger(ClientFactory.class);
+
+ /**
+ * Create the appropriate client intance based on configuration.
+ * @param config Spout configuration.
+ * @param instanceId Instance id of spout.
+ * @return Client.
+ */
+ public Client createClient(final RedisStreamSpoutConfig config, final int instanceId) {
+ Objects.requireNonNull(config);
+
+ switch (config.getClientType()) {
+ case JEDIS:
+ logger.info("Using Jedis client library.");
+ return new JedisClient(config, instanceId);
+ case LETTUCE:
+ logger.info("Using Lettuce client library.");
+ return new LettuceClient(config, instanceId);
+ default:
+ throw new IllegalStateException("Unknown/Unhandled Client Type");
+ }
+ }
+}

src/main/java/org/sourcelab/storm/spout/redis/client/ClientType.java

@@ -0,0 +1,9 @@
+package org.sourcelab.storm.spout.redis.client;
+
+/**
+ * Defines allowed implementations.
+ */
+public enum ClientType {
+ LETTUCE,
+ JEDIS;
+}

src/main/java/org/sourcelab/storm/spout/redis/client/jedis/JedisAdapter.java

@@ -0,0 +1,44 @@
+package org.sourcelab.storm.spout.redis.client.jedis;
+
+import redis.clients.jedis.StreamEntry;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Adapter to allow usage of both Jedis and JedisCluster.
+ */
+public interface JedisAdapter {
+ /**
+ * Call connect.
+ */
+ void connect();
+
+ /**
+ * Consume next batch of messages.
+ * @return List of messages consumed.
+ */
+ List<Map.Entry<String, List<StreamEntry>>> consume();
+
+ /**
+ * Mark the provided messageId as acknowledged/completed.
+ * @param msgId Id of the message.
+ */
+ void commit(final String msgId);
+
+ /**
+ * Disconnect client.
+ */
+ void close();
+
+ /**
+ * Advance the last offset consumed from PPL.
+ * @param lastMsgId Id of the last msg consumed.
+ */
+ void advancePplOffset(final String lastMsgId);
+
+ /**
+ * Switch to consuming from latest messages.
+ */
+ void switchToConsumerGroupMessages();
+}