Merge pull request #528 from alex268/master

Fixed lost compression errors

Author: alex268

topic/src/main/java/tech/ydb/topic/write/impl/EnqueuedMessage.java

@@ -1,6 +1,5 @@
 package tech.ydb.topic.write.impl;
 
-import java.io.IOException;
 import java.time.Instant;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -35,7 +34,7 @@ public class EnqueuedMessage {
     private final YdbTransaction transaction;
 
     private volatile boolean isReady = false;
-    private volatile IOException compressError = null;
+    private volatile Throwable compressError = null;
 
     public EnqueuedMessage(Message message, SendSettings sendSettings, boolean noCompression) {
         this.bytes = message.getData();
@@ -60,7 +59,7 @@ public long getSize() {
         return bytes.length;
     }
 
-    public IOException getCompressError() {
+    public Throwable getCompressError() {
         return compressError;
     }
 
@@ -71,8 +70,9 @@ public void encode(String writeId, Codec codec) {
             bytes = Encoder.encode(codec, bytes);
             isReady = true;
             logger.trace("[{}] Successfully finished encoding message", writeId);
-        } catch (IOException ex) {
+        } catch (Throwable ex) {
             logger.error("[{}] Exception while encoding message: ", writeId, ex);
+            compressError = ex;
             isReady = true;
             future.completeExceptionally(ex);
         }

topic/src/main/java/tech/ydb/topic/write/impl/WriterImpl.java

@@ -1,6 +1,5 @@
 package tech.ydb.topic.write.impl;
 
-import java.io.IOException;
 import java.util.Deque;
 import java.util.LinkedList;
 import java.util.List;
@@ -165,7 +164,7 @@ private void acceptMessageIntoSendingQueue(EnqueuedMessage message) {
         } else {
             CompletableFuture
                     .runAsync(() -> message.encode(id, settings.getCodec()), compressionExecutor)
-                    .thenRun(this::moveEncodedMessagesToSendingQueue);
+                    .whenComplete((res, th) -> moveEncodedMessagesToSendingQueue());
         }
     }
 
@@ -187,7 +186,7 @@ private void moveEncodedMessagesToSendingQueue() {
                     break;
                 }
 
-                IOException error = msg.getCompressError();
+                Throwable error = msg.getCompressError();
                 if (error != null) { // just skip
                     logger.warn("[{}] Message wasn't sent because of processing error", id, error);
                     free(1, msg.getOriginalSize());