Skip to content

fix(reader): implement retry logic for opening Redis reader #65

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

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

fix(reader): implement retry logic for opening Redis reader #65

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1beb697
fix(reader): implement retry logic for opening Redis reader
sauravkumarrr May 2, 2026
40f436d
fix: address review comments from Cursor Bugbot
sauravkumarrr May 2, 2026
b2265f5
fix(reader): ensure reader is closed on open failure
sauravkumarrr May 2, 2026
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
50 changes: 42 additions & 8 deletions ...redis-kafka-connect/src/main/java/com/redis/kafka/connect/source/RedisKeysSourceTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import io.lettuce.core.AbstractRedisClient;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.errors.RetriableException;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.springframework.batch.item.ExecutionContext;
Expand All @@ -30,8 +29,16 @@
import java.time.Clock;
import java.util.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RedisKeysSourceTask extends SourceTask {

private static final Logger log = LoggerFactory.getLogger(RedisKeysSourceTask.class);

static final int MAX_OPEN_RETRIES = 3;
static final long RETRY_DELAY_MS = 1000;

public static final Schema KEY_SCHEMA = Schema.STRING_SCHEMA;

private final ToStructFunction converter = new ToStructFunction();
Expand Down Expand Up @@ -77,11 +84,30 @@ public void start(Map<String, String> props) {
if (!config.getIdleTimeout().isNegative() && !config.getIdleTimeout().isZero()) {
reader.setIdleTimeout(config.getIdleTimeout());
}
try {
reader.open(new ExecutionContext());
} catch (ItemStreamException e) {
throw new RetriableException("Could not open reader", e);
openReaderWithRetry();
}

private void openReaderWithRetry() {
ItemStreamException lastException = null;
for (int attempt = 1; attempt <= MAX_OPEN_RETRIES; attempt++) {
try {
reader.open(new ExecutionContext());
return;
} catch (ItemStreamException e) {
lastException = e;
if (attempt < MAX_OPEN_RETRIES) {
log.warn("Failed to open reader (attempt {}/{}), retrying in {}ms",
attempt, MAX_OPEN_RETRIES, RETRY_DELAY_MS, e);
try {
Thread.sleep(RETRY_DELAY_MS);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new ConnectException("Interrupted while retrying reader open", e);
Comment thread
sauravkumarrr marked this conversation as resolved.
Outdated
}
}
}
}
throw new ConnectException("Could not open reader after " + MAX_OPEN_RETRIES + " attempts", lastException);
Comment thread
cursor[bot] marked this conversation as resolved.
}

@Deprecated
Expand All @@ -98,12 +124,20 @@ public void commit() throws InterruptedException {
@Override
public void stop() {
if (reader != null) {
reader.close();
try {
reader.close();
} catch (Exception e) {
log.warn("Error closing reader", e);
}
reader = null;
}
if (client != null) {
client.shutdown();
client.getResources().shutdown();
try {
client.shutdown();
client.getResources().shutdown();
} catch (Exception e) {
log.warn("Error shutting down Redis client", e);
}
Comment thread
sauravkumarrr marked this conversation as resolved.
client = null;
}
}
Expand Down
Loading