Addressing the encoding issues

Addressing some general cases

Author: tishun

src/main/java/io/lettuce/core/datastructure/queue/IncompleteQueue.java

@@ -76,10 +76,10 @@ private void encode(ChannelHandlerContext ctx, ByteBuf out, RedisCommand<?, ?, ?
         try {
             out.markWriterIndex();
             command.encode(out);
-        } catch (RuntimeException e) {
-            out.resetWriterIndex();
-            command.completeExceptionally(new EncoderException(
-                    "Cannot encode command. Please close the connection as the connection state may be out of sync.", e));
+        } catch (Throwable e) {
+            ctx.close();
+            logger.error("{} Cannot encode command. Closing the connection as the connection state may be out of sync.", logPrefix(ctx.channel()), e);
+            throw new EncoderException("Cannot encode command. Closing the connection as the connection state may be out of sync.", e);
         }
 
         if (debugEnabled) {

src/main/java/io/lettuce/core/protocol/CommandEncoder.java

@@ -46,7 +46,6 @@
 import io.lettuce.core.api.push.PushListener;
 import io.lettuce.core.api.push.PushMessage;
 import io.lettuce.core.datastructure.queue.HashIndexedQueue;
-import io.lettuce.core.datastructure.queue.IncompleteQueue;
 import io.lettuce.core.internal.LettuceAssert;
 import io.lettuce.core.internal.LettuceSets;
 import io.lettuce.core.metrics.CommandLatencyRecorder;
@@ -70,6 +69,7 @@
 import io.netty.util.internal.logging.InternalLogLevel;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
+import net.bytebuddy.implementation.bytecode.Throw;
 
 /**
  * A netty {@link ChannelHandler} responsible for writing redis commands and reading responses from the server.
@@ -161,11 +161,7 @@ public CommandHandler(ClientOptions clientOptions, ClientResources clientResourc
         this.commandLatencyRecorder = clientResources.commandLatencyRecorder();
         this.latencyMetricsEnabled = commandLatencyRecorder.isEnabled();
         this.boundedQueues = clientOptions.getRequestQueueSize() != Integer.MAX_VALUE;
-        Queue<RedisCommand<?, ?, ?>> implementation = clientOptions.isUseHashIndexedQueue() ? new HashIndexedQueue<>()
-                : new ArrayDeque<>();
-        this.stack = new IncompleteQueue(implementation);
-
-//        this.stack = clientOptions.isUseHashIndexedQueue() ? new HashIndexedQueue<>() : new ArrayDeque<>();
+        this.stack = clientOptions.isUseHashIndexedQueue() ? new HashIndexedQueue<>() : new ArrayDeque<>();
 
         Tracing tracing = clientResources.tracing();
 
@@ -664,8 +660,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Interrup
                         return;
                     }
 
-                } catch (Exception e) {
-
+                } catch (Throwable e) {
                     ctx.close();
                     throw e;
                 }
@@ -690,8 +685,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Interrup
                         decodeBufferPolicy.afterPartialDecode(buffer);
                         return;
                     }
-                } catch (Exception e) {
-
+                } catch (Throwable e) {
                     ctx.close();
                     throw e;
                 }
@@ -709,8 +703,9 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Interrup
                                 logger.debug("{} Completing command {}", logPrefix(), command);
                             }
                             complete(command);
-                        } catch (Exception e) {
+                        } catch (Throwable e) {
                             logger.warn("{} Unexpected exception during request: {}", logPrefix, e.toString(), e);
+                            command.completeExceptionally(e);
                         }
                     }
                 }