@@ -20,10 +20,11 @@
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public abstract class AbstractInputBuffer implements WorkflowInputBuffer {
-
   private final Collection<CustomObjectMarshaller> customMarshallers;
+  private final Map<Class<?>, CustomObjectMarshaller> marshallerMap = new ConcurrentHashMap<>();
 
   protected AbstractInputBuffer(Collection<CustomObjectMarshaller> customMarshallers) {
     this.customMarshallers = customMarshallers;
@@ -129,9 +130,12 @@ protected Class<?> loadClass(String className) throws ClassNotFoundException {
     return Class.forName(className);
   }
 
+  @SuppressWarnings("unchecked")
   protected Object readCustomObject() {
     Class<?> objectClass = readClass();
-    return MarshallingUtils.getCustomMarshaller(customMarshallers, objectClass)
+    return marshallerMap
+        .computeIfAbsent(
+            objectClass, o -> MarshallingUtils.getCustomMarshaller(customMarshallers, o))
         .read(this, objectClass);
   }
 }
@@ -18,10 +18,12 @@
 import java.time.Instant;
 import java.util.Collection;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 public abstract class AbstractOutputBuffer implements WorkflowOutputBuffer {
 
   private final Collection<CustomObjectMarshaller> customMarshallers;
+  private final Map<Class, CustomObjectMarshaller> marshallerMap = new ConcurrentHashMap<>();
 
   protected AbstractOutputBuffer(Collection<CustomObjectMarshaller> customMarshallers) {
     this.customMarshallers = customMarshallers;
@@ -111,9 +113,11 @@ protected void writeClass(Class<?> objectClass) {
 
   @SuppressWarnings({"rawtypes", "unchecked"})
   protected void writeCustomObject(Object object) {
+    Class<?> objectClass = object.getClass();
     CustomObjectMarshaller marshaller =
-        MarshallingUtils.getCustomMarshaller(customMarshallers, object.getClass());
-    writeClass(object.getClass());
+        marshallerMap.computeIfAbsent(
+            objectClass, o -> MarshallingUtils.getCustomMarshaller(customMarshallers, o));
+    writeClass(objectClass);
     marshaller.write(this, marshaller.getObjectClass().cast(object));
   }
 
 
@@ -18,13 +18,20 @@
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.lang.reflect.Modifier;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class MarshallingUtils {
 
+  private static final Logger logger = LoggerFactory.getLogger(MarshallingUtils.class);
+
   private MarshallingUtils() {}
 
   public static byte[] writeInstant(WorkflowBufferFactory factory, Instant instant) {
@@ -104,24 +111,75 @@ private static <T> T readValue(
    * class.
    *
    * @param marshallers Priority Sorted collection of marshalers available on classpath
-   * @param clazz The class of the object being marshaled
+   * @param objectClass The class of the object being marshaled
    * @return The most suitable marshaler for that object class
    * @throws IllegalArgumentException if no marshaler is found for that object class
    */
   @SuppressWarnings({"rawtypes", "unchecked"})
   public static CustomObjectMarshaller getCustomMarshaller(
-      Collection<CustomObjectMarshaller> marshallers, Class clazz) {
-    CustomObjectMarshaller assignable = null;
+      Collection<CustomObjectMarshaller> marshallers, Class objectClass) {
+    List<CustomObjectMarshaller> assignables = new ArrayList<>();
+    int minorClassDistance = Integer.MAX_VALUE;
+    int minorPriority = Integer.MAX_VALUE;
     for (CustomObjectMarshaller marshaller : marshallers) {
-      if (marshaller.getObjectClass().equals(clazz)) {
+      Class marshallerClass = marshaller.getObjectClass();
+      if (marshallerClass.equals(objectClass)) {
         return marshaller;
-      } else if (marshaller.getObjectClass().isAssignableFrom(clazz) && assignable == null) {
-        assignable = marshaller;
+      } else if (marshallerClass.isAssignableFrom(objectClass)) {
+        int classDistance = 0;
+        Class childClass = objectClass;
+        boolean found = false;
+        do {
+          classDistance++;
+          for (Class<?> candidate : childClass.getInterfaces()) {
+            if (candidate.equals(marshallerClass)) {
+              found = true;
+            }
+          }
+          if (!found) {
+            childClass = childClass.getSuperclass();
+            if (childClass == null) {
+              break;
+            } else {
+              found = childClass.equals(marshallerClass);
+            }
+          }
+        } while (!found && classDistance < minorClassDistance);
+        if (found) {
+          if (classDistance < minorClassDistance) {
+            assignables.clear();
+            assignables.add(marshaller);
+            minorClassDistance = classDistance;
+            minorPriority = marshaller.priority();
+          } else if (classDistance == minorClassDistance
+              && marshaller.priority() == minorPriority) {
+            assignables.add(marshaller);
+          }
+        }
       }
     }
-    if (assignable == null) {
-      throw new IllegalArgumentException("Cannot find proper marshaler for class " + clazz);
+    if (assignables.isEmpty()) {
+      throw new IllegalArgumentException("Cannot find proper marshaler for class " + objectClass);
+    }
+    return assignables.size() == 1 ? assignables.get(0) : chooseBetweenEquals(assignables);
+  }
+
+  @SuppressWarnings("rawtypes")
+  private static CustomObjectMarshaller chooseBetweenEquals(
+      List<CustomObjectMarshaller> marshallers) {
+    logger.debug("Several marshaling candidates with same priority {}", marshallers);
+    for (CustomObjectMarshaller marshaller : marshallers) {
+      Class marshallerClass = marshaller.getObjectClass();
+      if (!Modifier.isAbstract(marshallerClass.getModifiers())
+          && !marshallerClass.equals(Object.class)) {
+        return marshaller;
+      }
+    }
+    for (CustomObjectMarshaller marshaller : marshallers) {
+      if (Modifier.isAbstract(marshaller.getObjectClass().getModifiers())) {
+        return marshaller;
+      }
     }
-    return assignable;
+    return marshallers.get(0);
   }
 }
@@ -17,6 +17,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.io.Serializable;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -34,22 +35,46 @@ void testCustomMarshallers() {
     Mockito.when(employeeMarshaller.priority()).thenReturn(2);
     Mockito.when(employeeMarshaller.getObjectClass()).thenReturn(Employee.class);
     CustomObjectMarshaller objectMarshaller = Mockito.spy(CustomObjectMarshaller.class);
-    Mockito.when(objectMarshaller.priority()).thenReturn(3);
+    Mockito.when(objectMarshaller.priority()).thenReturn(1);
     Mockito.when(objectMarshaller.getObjectClass()).thenReturn(Object.class);
+    CustomObjectMarshaller ignoredMarshaller = Mockito.spy(CustomObjectMarshaller.class);
+    Mockito.when(ignoredMarshaller.priority()).thenReturn(2);
+    Mockito.when(ignoredMarshaller.getObjectClass()).thenReturn(Employee.class);
+    CustomObjectMarshaller serializableMarshaller = Mockito.spy(CustomObjectMarshaller.class);
+    Mockito.when(serializableMarshaller.priority()).thenReturn(1);
+    Mockito.when(serializableMarshaller.getObjectClass()).thenReturn(Serializable.class);
+
     Object employee = new Employee();
+    Object student = new Student();
     Object person = new Person();
-    Object other = new byte[2];
+    Object serializable = new SerializableClass();
+    Object other = new NonSerializableClass();
 
     List<CustomObjectMarshaller> marshallers =
-        Stream.of(objectMarshaller, employeeMarshaller, personMarshaller)
+        Stream.of(
+                objectMarshaller,
+                employeeMarshaller,
+                ignoredMarshaller,
+                personMarshaller,
+                serializableMarshaller)
             .sorted()
             .collect(Collectors.toList());
-    assertThat(marshallers).containsExactly(personMarshaller, employeeMarshaller, objectMarshaller);
+    assertThat(marshallers)
+        .containsExactly(
+            objectMarshaller,
+            personMarshaller,
+            serializableMarshaller,
+            employeeMarshaller,
+            ignoredMarshaller);
     assertThat(MarshallingUtils.getCustomMarshaller(marshallers, employee.getClass()))
         .isEqualTo(employeeMarshaller);
     assertThat(MarshallingUtils.getCustomMarshaller(marshallers, person.getClass()))
         .isEqualTo(personMarshaller);
     assertThat(MarshallingUtils.getCustomMarshaller(marshallers, other.getClass()))
         .isEqualTo(objectMarshaller);
+    assertThat(MarshallingUtils.getCustomMarshaller(marshallers, student.getClass()))
+        .isEqualTo(employeeMarshaller);
+    assertThat(MarshallingUtils.getCustomMarshaller(marshallers, serializable.getClass()))
+        .isEqualTo(serializableMarshaller);
   }
 }
@@ -0,0 +1,18 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.marshaller;
+
+class NonSerializableClass {}
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.marshaller;
+
+import java.io.Serializable;
+
+class SerializableClass implements Serializable {
+
+  private static final long serialVersionUID = 1L;
+}
@@ -0,0 +1,18 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.marshaller;
+
+class Student extends Employee {}