Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add Redis SET EXPIRE timeout argument, config param #25

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Add Redis SET EXPIRE timeout argument, config param #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e7b108a
Add Redis SET EXPIRE timeout argument, config param
kjocevicius May 30, 2023
1c3eede
Merge branch 'master' into feat/add_expire_param
kjocevicius Jun 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/sink/RedisSinkConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,10 @@ public enum PushDirection {
public static final String PUSH_DIRECTION_DOC = "List push direction: " + PushDirection.LEFT + " (LPUSH) or "
+ PushDirection.RIGHT + " (RPUSH)";

public static final String KEY_SET_EXPIRE_CONFIG = "redis.set.expire.timeout";
public static final String KEY_SET_EXPIRE_DEFAULT = "0";
public static final String KEY_SET_EXPIRE_DOC = "Key expiration timeout in millis for SET command.";

private final Charset charset;
private final DataType type;
private final String keyspace;
Expand All @@ -93,6 +97,7 @@ public enum PushDirection {
private final boolean multiexec;
private final int waitReplicas;
private final long waitTimeout;
private final long keySetExpireTimeout;

public RedisSinkConfig(Map<?, ?> originals) {
super(new RedisSinkConfigDef(), originals);
Expand All @@ -105,6 +110,7 @@ public RedisSinkConfig(Map<?, ?> originals) {
multiexec = Boolean.TRUE.equals(getBoolean(MULTIEXEC_CONFIG));
waitReplicas = getInt(WAIT_REPLICAS_CONFIG);
waitTimeout = getLong(WAIT_TIMEOUT_CONFIG);
keySetExpireTimeout = getLong(KEY_SET_EXPIRE_CONFIG);
}

public Charset getCharset() {
Expand Down Expand Up @@ -139,6 +145,10 @@ public long getWaitTimeout() {
return waitTimeout;
}

public long getKeySetExpireTimeout() {
return keySetExpireTimeout;
}

public static class RedisSinkConfigDef extends RedisConfigDef {

public RedisSinkConfigDef() {
Expand Down Expand Up @@ -168,6 +178,8 @@ private void define() {
.defaultValue(WAIT_REPLICAS_DEFAULT).importance(ConfigDef.Importance.MEDIUM).build());
define(ConfigKeyBuilder.of(WAIT_TIMEOUT_CONFIG, ConfigDef.Type.LONG).documentation(WAIT_TIMEOUT_DOC)
.defaultValue(WAIT_TIMEOUT_DEFAULT).importance(ConfigDef.Importance.MEDIUM).build());
define(ConfigKeyBuilder.of(KEY_SET_EXPIRE_CONFIG, ConfigDef.Type.LONG).documentation(KEY_SET_EXPIRE_DOC)
.defaultValue(KEY_SET_EXPIRE_DEFAULT).importance(ConfigDef.Importance.MEDIUM).build());
}

@Override
Expand Down
15 changes: 14 additions & 1 deletion core/redis-kafka-connect/src/main/java/com/redis/kafka/connect/sink/RedisSinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import io.lettuce.core.SetArgs;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigException;
Expand Down Expand Up @@ -81,6 +82,7 @@ public class RedisSinkTask extends SinkTask {
private RedisItemWriter<byte[], byte[], SinkRecord> writer;
private Converter jsonConverter;
private GenericObjectPool<StatefulConnection<byte[], byte[]>> pool;
private SetArgs setArgs;

@Override
public String version() {
Expand All @@ -97,6 +99,7 @@ public void start(final Map<String, String> props) {
jsonConverter.configure(Collections.singletonMap("schemas.enable", "false"), false);
writer = writer(client).options(config.writerOptions()).operation(operation());
writer.open(new ExecutionContext());
setArgs = getSetArgs(config);
final java.util.Set<TopicPartition> assignment = this.context.assignment();
if (!assignment.isEmpty()) {
Map<TopicPartition, Long> partitionOffsets = new HashMap<>(assignment.size());
Expand Down Expand Up @@ -146,7 +149,7 @@ private Operation<byte[], byte[], SinkRecord> operation() {
case JSON:
return JsonSet.key(this::key).value(this::jsonValue).del(this::isDelete).build();
case STRING:
return Set.key(this::key).value(this::value).del(this::isDelete).build();
return Set.key(this::key).value(this::value).args(setArgs).del(this::isDelete).build();
case STREAM:
return Xadd.key(this::collectionKey).body(this::map).build();
case LIST:
Expand Down Expand Up @@ -213,6 +216,16 @@ private boolean isDelete(SinkRecord sinkRecord) {
return sinkRecord.value() == null;
}

private SetArgs getSetArgs(RedisSinkConfig config) {
SetArgs setArgs = new SetArgs();

if (config.getKeySetExpireTimeout() > 0) {
setArgs.ex(config.getKeySetExpireTimeout());
}

return setArgs;
}

private byte[] key(SinkRecord sinkRecord) {
if (config.getKeyspace().isEmpty()) {
return bytes("key", sinkRecord.key());
Expand Down