https://flink.apache.org/ logo
Join Slack
Powered by
Hello, I am trying to create Redis async sink usi...
# troubleshooting
n
Hello, I am trying to create Redis async sink using lettuce and Flink's AsyncSinkBase . I ran the two jobs with same processing logic and resources but the sinks are different. One is Apache Bahir's implementation for Redis connector using Jedis library and extending RichSinkFunction. Another one is using lettuce and Flink's AsyncSinkBase. The flink pipeline with Lettuce and asyc implementation is not proceeding and showing very less numbers in the Records Sent of every operator. I am attaching the screenshots. Please help in identifying the issue. i can also share the async sink configuration in the comments. Lettuce library is supposed to be faster, seems like some issue with my configs. Additionally, i see the checkpoints are also failing for the lettuce pipeline. Any expert with Lettuce? Please help
d
Can you check the flamegraph for the busy sink and see what the bottleneck is?
πŸ‘ 1
n
@Danny Cranmer attached the screensshots of the flame graph. I coludnt really understand what is the bottleneck?
Copy code
public class RedisStateSerializer extends
        AsyncSinkWriterStateSerializer<Tuple3<String, String, Long>> {
    @Override
    protected void serializeRequestToStream(Tuple3<String, String, Long> request,
                                            DataOutputStream out) throws IOException {
        out.writeInt(request.f0.length());
        out.writeBytes(request.f0);
        out.writeInt(request.f1.length());
        out.writeBytes(request.f1);
        out.writeLong(request.f2);
    }

    @Override
    protected Tuple3<String, String, Long> deserializeRequestFromStream(long requestSize,
                                                                        DataInputStream in)
            throws IOException {

        return new Tuple3<>(readString(in), readString(in), in.readLong());
    }

    public String readString(DataInputStream dataInputStream) throws IOException {
        int length = dataInputStream.readInt();
        byte[] bytes = new byte[length];
        int bytesRead = 0;

        while (bytesRead < length) {
            int result = dataInputStream.read(bytes, bytesRead, length - bytesRead);
            if (result == -1) {
                throw new IOException("Premature end of stream.");
            }
            bytesRead += result;
        }

        return new String(bytes, StandardCharsets.UTF_8);
    }

    @Override
    public int getVersion() {
        return 1;
    }
}
Is this class the problem in any way?
d
The flame graph shows the sink is waiting for records to flush. This implies they are not flushing, or you are not handling errors correctly. Can you share your sink writer implementation?
n
Hello @Danny Cranmer, attaching the code for sink writer. In my testing, when i put single element in the source, it gets written to Redis successfully. Not sure why there is an issue and checkpoint/savepoints are failing.
Copy code
public class RedisSinkWriter<T> extends AsyncSinkWriter<T, Tuple3<String, String, Long>> {

    private static final ch.qos.logback.classic.Logger LOG =
            (ch.qos.logback.classic.Logger) LoggerFactory.getLogger("RollingAppender");
    private RedisURI redisUrl;
    private transient RedisClusterClient client = null;
    private transient StatefulRedisClusterConnection<String, String> clusterConnection = null;
    private transient RedisAdvancedClusterAsyncCommands<String, String> asyncCall = null;


    public RedisSinkWriter(RedisURI redisUrl,
                           ElementConverter<T, Tuple3<String, String, Long>> elementConverter,
                           Sink.InitContext context, int maxBatchSize, int maxInFlightRequests,
                           int maxBufferedRequests, long maxBatchSizeInBytes,
                           long maxTimeInBufferMS, long maxRecordSizeInBytes,
                           Collection<BufferedRequestState<Tuple3<String, String, Long>>>
                                   bufferedRequestStates) {
        super(elementConverter, context, maxBatchSize, maxInFlightRequests, maxBufferedRequests,
                maxBatchSizeInBytes, maxTimeInBufferMS, maxRecordSizeInBytes,
                bufferedRequestStates);
        this.redisUrl = redisUrl;
        if (client == null) {
            client = RedisClusterClient.create(redisUrl);
            ClusterTopologyRefreshOptions topologyRefreshOptions =
                    ClusterTopologyRefreshOptions.builder()
                    .enableAllAdaptiveRefreshTriggers()
                    .enablePeriodicRefresh(Duration.ofHours(1))
                    .dynamicRefreshSources(false)
                    .build();

            client.setOptions(ClusterClientOptions.builder().
                    topologyRefreshOptions(topologyRefreshOptions).build());
        }
        if (clusterConnection == null) {
            clusterConnection = client.connect();
        }
        if (asyncCall == null) {
            asyncCall  = clusterConnection.async();
        }

    }

    @Override
    protected void submitRequestEntries(List<Tuple3<String, String, Long>> requestEntries,
                                        Consumer<List<Tuple3<String, String, Long>>>
                                                requestToRetry) {
        for (Tuple3<String, String, Long> requestEntry : requestEntries) {
            if(requestEntry.f0!=null){
                RedisFuture<String> future = asyncCall.set(requestEntry.f0, requestEntry.f1);
                future.whenComplete((s, throwable) -> {
                    if (throwable != null) {
                        requestToRetry.accept(Collections.singletonList(requestEntry));
                    }
                });
            } else {
                LOG.error("redis.async.makeRequest: {} ", "Null key");
            }

        }
    }

    @Override
    protected long getSizeInBytes(Tuple3<String, String, Long> requestEntry) {
        return requestEntry.f0.length() * 2 +
                requestEntry.f1.length() * 2;
    }

    @Override
    public void close(){
        if (client != null) {
            client.shutdown();
        }
    }
}
d
In your 
submitRequestEntries
you're missing a 
requestEntriesToRetry(emptyList())
. When throwable is null. This signals the sink that the request is complete
This is why you are hanging on checkpoints, the sink does not register that in flight requests are complete, it is waiting for them to complete. There is no timeout there
n
It is working! thanks a lot @Danny Cranmer. I just put the throwable null check. I have another question. In my last operator, records sent is more than records received. How is it possible? Is there anything wrong with my above implementation?
d
Possibly due to retry? Are you getting any errors and retrying?
n
No Errors seen in the logs, i have put loggers in 'if (throwable != null) {' block but no errors are coming. Data is also going to Redis. I am not sure why the output metrics is greater than input.
2 Views
Open in Slack
PreviousNext
Powered by