Merge pull request #968: [proxima-beam-tools] register json objects to kryo

Author: je-ik

beam/tools/src/main/java/cz/o2/proxima/beam/tools/groovy/BeamStream.java

@@ -50,6 +50,11 @@
 import cz.o2.proxima.internal.com.google.common.base.Preconditions;
 import cz.o2.proxima.internal.com.google.common.collect.Iterables;
 import cz.o2.proxima.internal.com.google.common.collect.Streams;
+import cz.o2.proxima.internal.com.google.gson.JsonArray;
+import cz.o2.proxima.internal.com.google.gson.JsonElement;
+import cz.o2.proxima.internal.com.google.gson.JsonNull;
+import cz.o2.proxima.internal.com.google.gson.JsonObject;
+import cz.o2.proxima.internal.com.google.gson.JsonPrimitive;
 import cz.o2.proxima.tools.groovy.RepositoryProvider;
 import cz.o2.proxima.tools.groovy.Stream;
 import cz.o2.proxima.tools.groovy.StreamProvider.TerminatePredicate;
@@ -1259,6 +1264,11 @@ private static void registerCommonTypes(Kryo kryo, Repository repo) {
             java.time.Instant.class,
             Tuple.class,
             GStringImpl.class,
+            JsonElement.class,
+            JsonArray.class,
+            JsonNull.class,
+            JsonObject.class,
+            JsonPrimitive.class,
             String[].class,
             Integer[].class,
             Long[].class,

direct/io-bulkfs/src/main/java/cz/o2/proxima/direct/io/bulkfs/AbstractBulkFileSystemAttributeWriter.java

@@ -32,7 +32,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -79,7 +78,7 @@ synchronized void write(StreamElement data, CommitCallback commit, long watermar
     }
 
     public CommitCallback getCommit() {
-      return Objects.requireNonNull(commit);
+      return commit != null ? commit : CommitCallback.noop();
     }
   }