Improve ByteBufMapper.objectToBytes() contract

[controller.git] / atomix-storage / src / main / java / io / atomix / storage / journal / JournalSerdes.java

diff --git a/atomix-storage/src/main/java/io/atomix/storage/journal/JournalSerdes.java b/atomix-storage/src/main/java/io/atomix/storage/journal/JournalSerdes.java
index bf8518de2f8462bc77e83bd83ac64e1cd80e2e27..0265b8c6799e4bb2df144214d06fb3620b7616d2 100644 (file)
--- a/atomix-storage/src/main/java/io/atomix/storage/journal/JournalSerdes.java
+++ b/atomix-storage/src/main/java/io/atomix/storage/journal/JournalSerdes.java
@@ -21,6 +21,7 @@ import com.google.common.annotations.Beta;
 import com.google.common.annotations.VisibleForTesting;
 import io.atomix.utils.serializer.KryoJournalSerdesBuilder;
 import io.netty.buffer.ByteBuf;
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -125,7 +126,7 @@ public interface JournalSerdes {
                 try {
                     serialize(obj, buffer);
                 } catch (KryoException e) {
-                    throw new IOException(e);
+                    throw newIOException(e);
                 } finally {
                     // adjust writerIndex so that readableBytes() the bytes written
                     bytes.writerIndex(bytes.readerIndex() + buffer.position());
@@ -136,6 +137,24 @@ public interface JournalSerdes {
             public T bytesToObject(final long index, final ByteBuf bytes) {
                 return deserialize(bytes.nioBuffer());
             }
+
+            private static IOException newIOException(final KryoException cause) {
+                // We may have multiple nested KryoExceptions, intertwined with others, like IOExceptions. Let's find
+                // the deepest one.
+                var rootKryo = cause;
+                for (var nextCause = rootKryo.getCause(); nextCause != null; nextCause = nextCause.getCause()) {
+                    if (nextCause instanceof KryoException kryo) {
+                        rootKryo = kryo;
+                    }
+                }
+                // It would be nice to have a better way of discerning these, but alas it is what it is.
+                if (rootKryo.getMessage().startsWith("Buffer overflow.")) {
+                    final var ex = new EOFException();
+                    ex.initCause(cause);
+                    return ex;
+                }
+                return new IOException(rootKryo);
+            }
         };
     }