Merge pull request #674 from camunda-community-hub/enable-redis-clust…

…ers-with-connector-0.9.10

feat: upgrade to Redis exporter 0.9.10 and enable Redis cluster support

README.md

@@ -283,7 +283,9 @@ zeebe:
  worker:
  redis:
  connection: redis://localhost:6379
+ useClusterClient: false
  consumer-group: simple-monitor
+ prefix: zeebe
  xread-count: 500
  xread-block-millis: 2000
 
@@ -292,6 +294,9 @@ zeebe-importer: redis
 
 Refer to the [docker-compose file](docker/docker-compose.yml) for a sample configuration with the Redis importer. Profile presets: `redis,redis_in_memory`
 
+Please be aware that when connecting to a Redis cluster you must activate
+the `useClusterClient` option.
+
 ## Code of Conduct
 
 This project adheres to the Contributor Covenant [Code of

docker/docker-compose.yml

@@ -45,7 +45,8 @@ services:
 
  zeebe-redis:
  container_name: zeebe-broker-redis
- image: ghcr.io/camunda-community-hub/zeebe-with-redis-exporter:8.4.0
+ image: ghcr.io/camunda-community-hub/zeebe-with-redis-exporter:8.4.2
+ hostname: zeebe
  environment:
  - ZEEBE_REDIS_REMOTE_ADDRESS=redis://redis:6379
  - ZEEBE_REDIS_MAX_TIME_TO_LIVE_IN_SECONDS=900
@@ -135,7 +136,7 @@ services:
 
  simple-monitor-in-memory-redis:
  container_name: zeebe-simple-monitor-in-memory-redis
- image: ghcr.io/camunda-community-hub/zeebe-simple-monitor:2.6.4
+ image: ghcr.io/camunda-community-hub/zeebe-simple-monitor:2.7.1
  environment:
  - zeebe.client.broker.gateway-address=zeebe:26500
  - zeebe-importer=redis

pom.xml

@@ -21,7 +21,7 @@
  <zeebe.version>8.3.4</zeebe.version>
  <version.zeebe.spring>8.4.1</version.zeebe.spring>
  <hazelcast.exporter.version>1.4.0</hazelcast.exporter.version>
- <redis.exporter.version>0.9.9</redis.exporter.version>
+ <redis.exporter.version>0.9.10</redis.exporter.version>
 
  <querydsl.version>5.0.0</querydsl.version>
 

src/main/java/io/zeebe/monitor/config/RedisConfig.java

@@ -11,6 +11,9 @@ public class RedisConfig {
  @Value("${zeebe.client.worker.redis.connection}")
  private String redisConnection;
 
+ @Value("${zeebe.client.worker.redis.useClusterClient:false}")
+ private boolean useClusterClient;
+
  @Value("${zeebe.client.worker.redis.consumer-group:simple-monitor}")
  private String redisConumerGroup;
 
@@ -20,10 +23,17 @@ public class RedisConfig {
  @Value("${zeebe.client.worker.redis.xread-block-millis:2000}")
  private int redisXreadBlockMillis;
 
+ @Value("${zeebe.client.worker.redis.prefix:zeebe}")
+ private String redisPrefix;
+
  public String getRedisConnection() {
  return redisConnection;
  }
 
+ public boolean isUseClusterClient() {
+ return useClusterClient;
+ }
+
  public String getRedisConumerGroup() {
  return redisConumerGroup;
  }
@@ -35,4 +45,8 @@ public int getRedisXreadCount() {
  public int getRedisXreadBlockMillis() {
  return redisXreadBlockMillis;
  }
+
+ public String getRedisPrefix() {
+ return redisPrefix;
+ }
 }

src/main/java/io/zeebe/monitor/zeebe/redis/RedisImportService.java

@@ -1,13 +1,11 @@
 package io.zeebe.monitor.zeebe.redis;
 
-import com.hazelcast.core.HazelcastInstance;
 import io.lettuce.core.RedisClient;
+import io.lettuce.core.cluster.RedisClusterClient;
 import io.zeebe.exporter.proto.Schema;
-import io.zeebe.hazelcast.connect.java.ZeebeHazelcast;
 import io.zeebe.monitor.config.RedisConfig;
-import io.zeebe.monitor.entity.HazelcastConfig;
-import io.zeebe.monitor.repository.HazelcastConfigRepository;
 import io.zeebe.monitor.zeebe.protobuf.importers.*;
+import io.zeebe.redis.connect.java.RedisConnectionBuilder;
 import io.zeebe.redis.connect.java.ZeebeRedis;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -28,44 +26,57 @@ public class RedisImportService {
  @Autowired private ErrorProtobufImporter errorImporter;
 
  public ZeebeRedis importFrom(final RedisClient redisClient, RedisConfig redisConfig) {
+ final var builder = ZeebeRedis.newBuilder(redisClient)
+ .consumerGroup(redisConfig.getRedisConumerGroup())
+ .xreadCount(redisConfig.getRedisXreadCount()).xreadBlockMillis(redisConfig.getRedisXreadBlockMillis())
+ .prefix(redisConfig.getRedisPrefix());
+ addListener(builder);
+ return builder.build();
+ }
 
- final var builder =
- ZeebeRedis.newBuilder(redisClient)
- .consumerGroup(redisConfig.getRedisConumerGroup())
- .xreadCount(redisConfig.getRedisXreadCount()).xreadBlockMillis(redisConfig.getRedisXreadBlockMillis())
- .addProcessListener(
- record -> ifEvent(record, Schema.ProcessRecord::getMetadata, processAndElementImporter::importProcess))
+ public ZeebeRedis importFrom(final RedisClusterClient redisClient, RedisConfig redisConfig) {
+ final var builder = ZeebeRedis.newBuilder(redisClient)
+ .withStandardClusterOptions()
+ .consumerGroup(redisConfig.getRedisConumerGroup())
+ .xreadCount(redisConfig.getRedisXreadCount()).xreadBlockMillis(redisConfig.getRedisXreadBlockMillis())
+ .prefix(redisConfig.getRedisPrefix());
+ addListener(builder);
+ return builder.build();
+ }
+
+ private void addListener(RedisConnectionBuilder connectionBuilder) {
+ connectionBuilder
+ .addProcessListener(record ->
+ ifEvent(record, Schema.ProcessRecord::getMetadata, processAndElementImporter::importProcess))
  .addProcessInstanceListener(
- record ->
- ifEvent(
- record,
- Schema.ProcessInstanceRecord::getMetadata,
- processAndElementImporter::importProcessInstance))
+  record ->
+  ifEvent(
+  record,
+  Schema.ProcessInstanceRecord::getMetadata,
+  processAndElementImporter::importProcessInstance))
  .addIncidentListener(
- record -> ifEvent(record, Schema.IncidentRecord::getMetadata, incidentImporter::importIncident))
+  record -> ifEvent(record, Schema.IncidentRecord::getMetadata, incidentImporter::importIncident))
  .addJobListener(
- record -> ifEvent(record, Schema.JobRecord::getMetadata, jobImporter::importJob))
+  record -> ifEvent(record, Schema.JobRecord::getMetadata, jobImporter::importJob))
  .addVariableListener(
- record -> ifEvent(record, Schema.VariableRecord::getMetadata, variableImporter::importVariable))
+  record -> ifEvent(record, Schema.VariableRecord::getMetadata, variableImporter::importVariable))
  .addTimerListener(
- record -> ifEvent(record, Schema.TimerRecord::getMetadata, timerImporter::importTimer))
+  record -> ifEvent(record, Schema.TimerRecord::getMetadata, timerImporter::importTimer))
  .addMessageListener(
- record -> ifEvent(record, Schema.MessageRecord::getMetadata, messageImporter::importMessage))
+  record -> ifEvent(record, Schema.MessageRecord::getMetadata, messageImporter::importMessage))
  .addMessageSubscriptionListener(
- record ->
- ifEvent(
- record,
- Schema.MessageSubscriptionRecord::getMetadata,
- messageSubscriptionImporter::importMessageSubscription))
+  record ->
+  ifEvent(
+  record,
+  Schema.MessageSubscriptionRecord::getMetadata,
+  messageSubscriptionImporter::importMessageSubscription))
  .addMessageStartEventSubscriptionListener(
- record ->
- ifEvent(
- record,
- Schema.MessageStartEventSubscriptionRecord::getMetadata,
- messageSubscriptionImporter::importMessageStartEventSubscription))
+  record ->
+  ifEvent(
+  record,
+  Schema.MessageStartEventSubscriptionRecord::getMetadata,
+  messageSubscriptionImporter::importMessageStartEventSubscription))
  .addErrorListener(errorImporter::importError);
-
- return builder.build();
  }
 
  private <T> void ifEvent(

src/main/java/io/zeebe/monitor/zeebe/redis/ZeebeRedisService.java

@@ -2,6 +2,7 @@
 
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisURI;
+import io.lettuce.core.cluster.RedisClusterClient;
 import io.zeebe.monitor.config.RedisConfig;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
@@ -34,10 +35,15 @@ public void start() {
  var redisUri = RedisURI.create(config.getRedisConnection());
 
  LOG.info("Connecting to Redis {}, consumer group {}", redisUri, config.getRedisConumerGroup());
- var redisClient = RedisClient.create(redisUri);
-
- LOG.info("Importing records from Redis...");
- closeable = importService.importFrom(redisClient, config);
+ if (config.isUseClusterClient()) {
+ var redisClient = RedisClusterClient.create(redisUri);
+ LOG.info("Importing records from Redis cluster...");
+ closeable = importService.importFrom(redisClient, config);
+ } else {
+ var redisClient = RedisClient.create(redisUri);
+ LOG.info("Importing records from Redis...");
+ closeable = importService.importFrom(redisClient, config);
+ }
  }
 
  @PreDestroy