Add Support for Key TTL Setting for Sink Connector

Dockerfile

@@ -1,6 +1,6 @@
-ARG CP_VERSION=7.2.0
+ARG CP_VERSION=7.7.0
 ARG BASE_PREFIX=confluentinc
-ARG CONNECT_IMAGE=cp-server-connect
+ARG CONNECT_IMAGE=cp-kafka-connect
 
 FROM openjdk:18-jdk-slim AS build
 WORKDIR /root/redis-kafka-connect

core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/sink/RedisSinkConfig.java

@@ -37,6 +37,7 @@ public enum RedisType {
 	private final boolean multiExec;
 	private final int waitReplicas;
 	private final Duration waitTimeout;
+	private final long keyTTL;
 
 	public RedisSinkConfig(Map<?, ?> originals) {
 		super(new RedisSinkConfigDef(), originals);
@@ -48,6 +49,7 @@ public RedisSinkConfig(Map<?, ?> originals) {
 		multiExec = Boolean.TRUE.equals(getBoolean(RedisSinkConfigDef.MULTIEXEC_CONFIG));
 		waitReplicas = getInt(RedisSinkConfigDef.WAIT_REPLICAS_CONFIG);
 		waitTimeout = Duration.ofMillis(getLong(RedisSinkConfigDef.WAIT_TIMEOUT_CONFIG));
+		keyTTL = getLong(RedisSinkConfigDef.KEY_TTL_CONFIG);
 	}
 
 	public Charset getCharset() {
@@ -78,12 +80,16 @@ public Duration getWaitTimeout() {
 		return waitTimeout;
 	}
 
+	public long getKeyTTL() {
+		return keyTTL;
+	}
+
 	@Override
 	public int hashCode() {
 		final int prime = 31;
 		int result = super.hashCode();
 		result = prime * result
-				+ Objects.hash(charset, keyspace, separator, multiExec, type, waitReplicas, waitTimeout);
+				+ Objects.hash(charset, keyspace, separator, multiExec, type, waitReplicas, waitTimeout, keyTTL);
 		return result;
 	}
 
@@ -98,7 +104,7 @@ public boolean equals(Object obj) {
 		RedisSinkConfig other = (RedisSinkConfig) obj;
 		return Objects.equals(charset, other.charset) && Objects.equals(keyspace, other.keyspace)
 				&& Objects.equals(separator, other.separator) && multiExec == other.multiExec && type == other.type
-				&& waitReplicas == other.waitReplicas && waitTimeout == other.waitTimeout;
+				&& waitReplicas == other.waitReplicas && waitTimeout == other.waitTimeout && keyTTL == other.keyTTL;
 	}
 
 }

core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/sink/RedisSinkConfigDef.java

@@ -61,6 +61,12 @@ public class RedisSinkConfigDef extends RedisConfigDef {
 	public static final String TYPE_DOC = "Destination data structure: "
 			+ String.join(",", Stream.of(RedisType.values()).map(RedisType::name).toArray(String[]::new));
 
+	public static final String KEY_TTL_CONFIG = "redis.key.ttl";
+
+	public static final String KEY_TTL_CONFIG_DEFAULT = "-1";
+
+	public static final String KEY_TTL_CONFIG_DOC = "Time to live in seconds for the key. If not set, the record will not expire.";
+
 	protected static final Set<RedisType> MULTI_EXEC_COMMANDS = Stream
 			.of(RedisType.STREAM, RedisType.LIST, RedisType.SET, RedisType.ZSET).collect(Collectors.toSet());
 
@@ -81,6 +87,7 @@ private void define() {
 		define(MULTIEXEC_CONFIG, Type.BOOLEAN, MULTIEXEC_DEFAULT, Importance.MEDIUM, MULTIEXEC_DOC);
 		define(WAIT_REPLICAS_CONFIG, Type.INT, WAIT_REPLICAS_DEFAULT, Importance.MEDIUM, WAIT_REPLICAS_DOC);
 		define(WAIT_TIMEOUT_CONFIG, Type.LONG, WAIT_TIMEOUT_DEFAULT, Importance.MEDIUM, WAIT_TIMEOUT_DOC);
+		define(KEY_TTL_CONFIG, Type.LONG, KEY_TTL_CONFIG_DEFAULT, Importance.MEDIUM, KEY_TTL_CONFIG_DOC);
 	}
 
 	@Override