Do not requeue commands on unrecoverable errors #680

Commands are canceled now when the attempted write fails with an recoverable error (EncoderException, Errors).
redis · Jan 30, 2018 · cb1bda3 · cb1bda3
1 parent 9f47a2c
commit cb1bda3
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 4 deletions.
diff --git a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
@@ -31,6 +31,7 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
+import io.netty.handler.codec.EncoderException;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.internal.logging.InternalLogLevel;
@@ -734,6 +735,12 @@ public void operationComplete(Future<Void> future) throws Exception {
             dequeue();
 
             if (!success) {
+
+                if (cause instanceof EncoderException || cause instanceof Error || cause.getCause() instanceof Error) {
+                    complete(cause);
+                    return;
+                }
+
                 Channel channel = DefaultEndpoint.this.channel;
                 if (channel != null) {
                     channel.eventLoop().submit(this::requeueCommands);

diff --git a/src/test/java/io/lettuce/core/protocol/DefaultEndpointTest.java b/src/test/java/io/lettuce/core/protocol/DefaultEndpointTest.java
@@ -50,9 +50,10 @@
 import io.lettuce.core.codec.Utf8StringCodec;
 import io.lettuce.core.internal.LettuceFactories;
 import io.lettuce.core.output.StatusOutput;
-import io.lettuce.core.resource.ClientResources;
 import io.netty.channel.Channel;
 import io.netty.channel.DefaultChannelPromise;
+import io.netty.handler.codec.EncoderException;
+import io.netty.util.concurrent.ImmediateEventExecutor;
 
 @RunWith(MockitoJUnitRunner.class)
 public class DefaultEndpointTest {
@@ -70,9 +71,6 @@ public class DefaultEndpointTest {
     @Mock
     private ConnectionFacade connectionFacade;
 
-    @Mock
-    private ClientResources clientResources;
-
     @Mock
     private ConnectionWatchdog connectionWatchdog;
 
@@ -203,6 +201,32 @@ public void notifyDrainQueuedCommandsShouldWriteCommands() {
         verify(channel).writeAndFlush(eq(Arrays.asList(command)));
     }
 
+    @Test
+    public void shouldCancelCommandsOnEncoderException() {
+
+        when(channel.isActive()).thenReturn(true);
+        sut.notifyChannelActive(channel);
+
+        DefaultChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
+
+        when(channel.writeAndFlush(any())).thenAnswer(invocation -> {
+            if (invocation.getArguments()[0] instanceof RedisCommand) {
+                queue.add((RedisCommand) invocation.getArguments()[0]);
+            }
+
+            if (invocation.getArguments()[0] instanceof Collection) {
+                queue.addAll((Collection) invocation.getArguments()[0]);
+            }
+            return promise;
+        });
+
+        promise.setFailure(new EncoderException("foo"));
+
+        sut.write(command);
+
+        assertThat(command.exception).isInstanceOf(EncoderException.class);
+    }
+
     @Test
     public void writeShouldRejectCommandsInDisconnectedState() {