Should ReactiveCommandDispatcher.java check subscription status? #323
Comments
|
Hi @vleushin, |
|
I also recall that Glad to help. Second question -- how bad it is for |
|
You can also unsubscribe during emission of items, so having that safeguards in place isn't too bad of an idea. Lettuce should not get into an invalid state. Data streaming using Calls to |
Perform a check to isUnsubscribed() before emitting items or calling completion methods
Perform a check to isUnsubscribed() before emitting items or calling completion methods
|
I added the check to
|
|
@vleushin I had a chat with the RxJava guys. Throwing exceptions in |
|
This is interesting. Thanks! I think I must tell that to GRPC guys. On the other hand -- they are throwing it in case when you should not call in (when it is unsubscribed). |
|
I added the fix to check for |
|
Thanks. Meanwhile, I tried new library setup on production -- and I got same problem. Getting completely wrong results. Say, user 1 does I think I understand how to reproduce it on my local machine and I will be debugging this tomorrow. I'll share my findings. |
|
Thanks a lot for your feedback. I'd also like to understand the issue and get it fixed. |
|
@mp911de I finally managed to pinpoint it. It is definitely because of that exception in Here is simple test to reproduce problem: @Test
public void testWrongValue() throws ExecutionException, InterruptedException {
final RedisClient client = RedisClient.create("redis://localhost/0");
final StatefulRedisConnection<String, String> connect = client.connect();
final RedisReactiveCommands<String, String> commands = connect.reactive();
Observable.concat(
commands.set("keyA", "valueA"),
commands.set("keyB", "valueB"))
.lastOrDefault(null)
.toBlocking()
.toFuture()
.get();
commands.get("keyA").subscribe(createSubscriberWithExceptionOnComplete());
// bonus: comment this line and it will work fine
commands.get("keyA").subscribe(createSubscriberWithExceptionOnComplete());
Thread.sleep(100);
final String valueB = commands.get("keyB")
.toBlocking()
.toFuture()
.get();
System.out.println(valueB); // prints valueA
}
private static Subscriber<String> createSubscriberWithExceptionOnComplete() {
return new Subscriber<String>() {
@Override
public void onCompleted() {
throw new RuntimeException("throwing something");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
}
};
} |
|
I inspected protected void decode(ChannelHandlerContext ctx, ByteBuf buffer) throws InterruptedException {
while (!queue.isEmpty()) {
RedisCommand<K, V, ?> command = queue.peek();
if (debugEnabled) {
logger.debug("{} Queue contains: {} commands", logPrefix(), queue.size());
}
WithLatency withLatency = null;
if (clientResources.commandLatencyCollector().isEnabled()) {
RedisCommand<K, V, ?> unwrappedCommand = CommandWrapper.unwrap(command);
if (unwrappedCommand instanceof WithLatency) {
withLatency = (WithLatency) unwrappedCommand;
if (withLatency.getFirstResponse() == -1) {
withLatency.firstResponse(nanoTime());
}
}
}
if (!rsm.decode(buffer, command, command.getOutput())) {
return;
}
command = queue.poll();
recordLatency(withLatency, command.getType());
command.complete();
if (buffer != null && buffer.refCnt() != 0) {
buffer.discardReadBytes();
}
}
}And I think because of that exception in One more observation that let me think about commands getting shifted: final String firstValueB = commands.get("keyB")
.toBlocking()
.toFuture()
.get();
final String secondValueB = commands.get("keyB")
.toBlocking()
.toFuture()
.get();
System.out.println(firstValueB); // prints valueA
System.out.println(secondValueB); // prints valueB |
|
@mp911de What do you think? I tried wrapping |
|
Thanks for digging into the issue. I need to get up to speed. Was on travel last week and got sick, but I'll come back to you. |
|
I took a look at the issue. Your fix will work but that's not the root cause. The buffer maintains a reader index so additional incoming bytes can resume from the position where decoding was finished the last time bytes were received. That code is safe. Another issue causes the problem: The exception is propagated to the exception handler and
I will apply your suggested fix. |
Command completion is now guarded by try/catch to prevent the global exception handler to kick in. Commands must be removed from the queue prior to completion to prevent duplicate submission on ping-before-reconnect. See also #323
Command completion is now guarded by try/catch to prevent the global exception handler to kick in. Commands must be removed from the queue prior to completion to prevent duplicate submission on ping-before-reconnect. See also #323
|
I'm very glad we figured that out! Thanks! |
Hello, I'm back to upgrading libraries and my issue #202 is probably still there. I tried on staging environment and I get this exception:
Let's say it is ok.
Later I decided to check https://github.com/mp911de/lettuce/blob/master/src/main/java/com/lambdaworks/redis/ReactiveCommandDispatcher.java
And I noticed that you do not check the subscription status (
isUnsubscribed). So I think that I understand the problem now: I'm doing some logic in my command and sending some requests to Redis, command gets cancelled because of timeout or whatever reason,lettucegets results and notifies unsubscribed subscriber about it (onNext,onCompleted,onError). and gets exception that command was cancelled.So the question goes lis: should we check the subscription status before notifying subscriber about result or not. As far as I know we should, but I have to double check Rx docs to be sure.
The text was updated successfully, but these errors were encountered: