Do not use ByteBufferInput

ByteBufferInput ends up calling ByteBuffer.put() in readAscii(). Work
this around by integrating through InputStream. Not exactly nice, but
the best we can do right now.

JIRA: CONTROLLER-2109
Change-Id: I182c3cefb97ada0cd9fdc2b7440256f38341df5a
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>

diff --git a/atomix-storage/src/main/java/io/atomix/utils/serializer/KryoJournalSerdes.java b/atomix-storage/src/main/java/io/atomix/utils/serializer/KryoJournalSerdes.java
index d99c69d9b6829f76894bf9ffe39b5bcbf88f7109..f3215e235889d725d4c87491743671a8c2b73bfc 100644 (file)
--- a/atomix-storage/src/main/java/io/atomix/utils/serializer/KryoJournalSerdes.java
+++ b/atomix-storage/src/main/java/io/atomix/utils/serializer/KryoJournalSerdes.java
@@ -22,6 +22,7 @@ import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Registration;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.ByteBufferInput;
+import com.esotericsoftware.kryo.io.ByteBufferInputStream;
 import com.esotericsoftware.kryo.io.ByteBufferOutput;
 import com.esotericsoftware.kryo.pool.KryoCallback;
 import com.esotericsoftware.kryo.pool.KryoFactory;
@@ -143,15 +144,14 @@ final class KryoJournalSerdes implements JournalSerdes, KryoFactory, KryoPool {
 
     @Override
     public <T> T deserialize(final ByteBuffer buffer) {
-        ByteBufferInput in = new ByteBufferInput(buffer);
-        Kryo kryo = borrow();
-        try {
-            @SuppressWarnings("unchecked")
-            T obj = (T) kryo.readClassAndObject(in);
-            return obj;
-        } finally {
-            release(kryo);
-        }
+        return kryoInputPool.run(input -> {
+            input.setInputStream(new ByteBufferInputStream(buffer));
+            return kryoPool.run(kryo -> {
+                @SuppressWarnings("unchecked")
+                T obj = (T) kryo.readClassAndObject(input);
+                return obj;
+            });
+        }, DEFAULT_BUFFER_SIZE);
     }
 
     @Override
@@ -161,11 +161,10 @@ final class KryoJournalSerdes implements JournalSerdes, KryoFactory, KryoPool {
 
     @Override
     public <T> T deserialize(final InputStream stream, final int bufferSize) {
-        ByteBufferInput in = new ByteBufferInput(stream, bufferSize);
         Kryo kryo = borrow();
         try {
             @SuppressWarnings("unchecked")
-            T obj = (T) kryo.readClassAndObject(in);
+            T obj = (T) kryo.readClassAndObject(new ByteBufferInput(stream, bufferSize));
             return obj;
         } finally {
             release(kryo);