diff --git a/pom.xml b/pom.xml
index 468ecd88e..d189a7df5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -100,6 +100,12 @@
 			<optional>true</optional>
 		</dependency>
 
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+
 		<!--  RxJava -->
 
 		<dependency>
diff --git a/src/main/asciidoc/entity-callbacks.adoc b/src/main/asciidoc/entity-callbacks.adoc
new file mode 100644
index 000000000..bd707ad7c
--- /dev/null
+++ b/src/main/asciidoc/entity-callbacks.adoc
@@ -0,0 +1,156 @@
+[entity-callbacks]
+= Entity Callbacks
+
+The Spring Data infrastructure provides hooks for modifying an entity before and/or after certain methods are invoked.
+Those so called ``EntityCallback``s provide a convenient way to check and potentially modify an entity in a callback fashioned style. +
+An `EntityCallback` looks pretty much like a specialized `ApplicationListener`, with which you might be familiar.
+Some Spring Data implementations publish store specific events, like a `BeforeSaveEvent` that allow to modify the given entity, which in some cases, eg. immutable types, can cause somme trouble.
+Plus, the event publishing relies on the `ApplicationEventMulticaster` which can be configured with an asynchronous `TaskExecutor` leading to unpredictable outcome.
+
+``EntityCallback``s offer integration points with both sync and reactive APIs guaranteeing an in order invocation at fixed checkpoints within the processing chain, returning a potentially modified entity or an reactive wrapper type.
+
+[NOTE]
+====
+The entity callback API has been introduced with Spring Data Commons 2.2 and is the recommended way of dealing with entity modifications. +
+Existing store specific `ApplicationEvents` are still be published *before* the potentially registered ``EntityCallback``s are called.
+====
+
+[entity-callbacks.implement]
+== Implementing Entity Callbacks
+
+The `EntityCallback` is, via its generic type argument, directly associated with the domain type it is meant for.
+Each store typically uses a set of predefined entity callbacks covering the lifecycle of an entity.
+
+.Anatomy of an `EntityCallback`
+====
+[source,java]
+----
+@FunctionalInterface
+public interface BeforeSaveCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked before a domain object is saved.
+	 * Can return either the same or a modified instance.
+	 *
+	 * @return the domain object to be persisted.
+	 */
+	T onBeforeSave(T entity <2>, String collection <3>); <1>
+}
+----
+<1> `BeforeSaveCallback` specific method to be called before an entity is saved. Returns a potentially modifed instance.
+<2> The entity right bevore presisted.
+<3> A number of store specific arguments like the _collection_ the entity is persisted to.
+====
+
+.Anatomy of a reactive `EntityCallback`
+====
+[source,java]
+----
+@FunctionalInterface
+public interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked on subscription, before a domain object is saved.
+	 * The returned Publisher can emit either the same or a modified instance.
+	 *
+	 * @return Publisher emitting the domain object to be persisted.
+	 */
+	Publisher<T> onBeforeSave(T entity <2>, String collection <3>); <1>
+}
+----
+<1> `BeforeSaveCallback` specific method to be called on subscription, before an entity is saved. Emits a potentially modifed instance.
+<2> The entity right bevore presisted.
+<3> A number of store specific arguments like the _collection_ the entity is persisted to.
+====
+
+Implement the interface suiting your application needs like shown in the example below.
+
+.Example `BeforeSaveCallback`
+====
+[source,java]
+----
+public class DefaultingEntityCallback implements BeforeSaveCallback<Person>, Ordered <2> {
+
+	@Override
+	public Object onBeforeSave(Person entity, String collection) { <1>
+
+		if(collection == "user) {
+		    return // ...
+		}
+
+		return // ...
+	}
+
+	@Override
+	public int getOrder() {
+		return 100; <2>
+	}
+}
+----
+<1> Implement logic according to application requirements.
+<2> Potentially order the entity callback if multiple ones for the same domain type exist. Ordering follows lowest precedence.
+====
+
+[entity-callbacks.register]
+== Registering Entity Callbacks
+
+``EntityCallback``s get picked up by the store specific implementations in case they are provided with an `ApplicationContext`.
+Most template APIs already implement `ApplicationContextAware` an therefore have a context at hand if registered as a Bean.
+
+The following example provides a collection of valid entity callback registrations.
+
+.Example `EntityCallback` Bean registration
+====
+[source,java]
+----
+@Configuration
+public class EntityCallbackConfiguration {
+
+    @Bean
+    BeforeSaveCallback<Person> annotationOrderedCallback() { <1>
+    	return new First();
+    }
+
+    @Bean
+    BeforeSaveCallback<Person> interfaceOrderedCallback() { <2>
+    	return new DefaultingEntityCallback();
+    }
+
+    @Bean
+    BeforeSaveCallback<Person> unorderedLambdaReceiverCallback() { <3>
+        return (BeforeSaveCallback<Person>) it -> // ...
+    }
+
+    @Bean
+    UserCallbacks multipleCallbacksInOneImplementationClass() { <4>
+        return new UserCallbacks();
+    }
+
+    @Order(1) <1>
+    static class First implements BeforeSaveCallback<Person> {
+
+        @Override
+    	public Person onBeforeSave(Person person) {
+    		return // ...
+    	}
+    }
+
+    static class UserCallbacks implements BeforeConvertCallback<User>, BeforeSaveCallback<User> { <4>
+
+        @Override
+        public Person onBeforeConvert(User user) {
+        	return // ...
+        }
+
+        @Override
+        public Person onBeforeSave(User user) {
+        	return // ...
+        }
+    }
+}
+----
+<1> `BeforeSaveCallback` receiving its order from the `@Order` annotation.
+<2> `BeforeSaveCallback` receiving its order via the `Ordered` interface implementation.
+<3> `BeforeSaveCallback` using a lambda expression. Unordered by default and invoked last.
+<4> Combine multiple entity callback interfaces in one implementation class.
+====
diff --git a/src/main/java/org/springframework/data/mapping/callback/DefaultEntityCallbacks.java b/src/main/java/org/springframework/data/mapping/callback/DefaultEntityCallbacks.java
new file mode 100644
index 000000000..a5ec6718b
--- /dev/null
+++ b/src/main/java/org/springframework/data/mapping/callback/DefaultEntityCallbacks.java
@@ -0,0 +1,138 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.core.ResolvableType;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * {@link EntityCallbacks} implementation using an {@link EntityCallbackDiscoverer} to retrieve {@link EntityCallback
+ * EntityCallbacks} from a {@link BeanFactory}.
+ * 
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+class DefaultEntityCallbacks implements EntityCallbacks {
+
+	private final Map<Class<?>, Method> callbackMethodCache = new ConcurrentReferenceHashMap<>(64);
+	private final SimpleEntityCallbackInvoker callbackInvoker = new SimpleEntityCallbackInvoker();
+	private final EntityCallbackDiscoverer callbackDiscoverer;
+
+	/**
+	 * Create new instance of {@link DefaultEntityCallbacks}.
+	 */
+	DefaultEntityCallbacks() {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer();
+	}
+
+	/**
+	 * Create new instance of {@link DefaultEntityCallbacks} discovering {@link EntityCallback entity callbacks} within
+	 * the given {@link BeanFactory}.
+	 * 
+	 * @param beanFactory must not be {@literal null}.
+	 */
+	DefaultEntityCallbacks(BeanFactory beanFactory) {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer(beanFactory);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.EntityCallbacks#callback(java.lang.Class, java.lang.Object, java.lang.Object)
+	 */
+	@Override
+	public <T> T callback(Class<? extends EntityCallback> callbackType, T entity, Object... args) {
+
+		Assert.notNull(entity, "Entity must not be null!");
+
+		Class<T> entityType = (Class<T>) (entity != null ? ClassUtils.getUserClass(entity.getClass())
+				: callbackDiscoverer.resolveDeclaredEntityType(callbackType).getRawClass());
+
+		Method callbackMethod = callbackMethodCache.computeIfAbsent(callbackType, it -> {
+
+			Method method = EntityCallbackDiscoverer.lookupCallbackMethod(it, entityType, args);
+			ReflectionUtils.makeAccessible(method);
+			return method;
+		});
+
+		T value = entity;
+
+		for (EntityCallback<T> callback : callbackDiscoverer.getEntityCallbacks(entityType,
+				ResolvableType.forClass(callbackType))) {
+
+			BiFunction<EntityCallback<T>, T, Object> callbackFunction = EntityCallbackDiscoverer
+					.computeCallbackInvokerFunction(callback, callbackMethod, args);
+			value = callbackInvoker.invokeCallback(callback, value, callbackFunction);
+		}
+
+		return value;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.EntityCallbacks#addEntityCallback(org.springframework.data.mapping.callback.EntityCallback)
+	 */
+	@Override
+	public void addEntityCallback(EntityCallback<?> callback) {
+		this.callbackDiscoverer.addEntityCallback(callback);
+	}
+
+	class SimpleEntityCallbackInvoker implements org.springframework.data.mapping.callback.EntityCallbackInvoker {
+
+		@Override
+		public <T> T invokeCallback(EntityCallback<T> callback, T entity,
+				BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction) {
+
+			try {
+
+				Object value = callbackInvokerFunction.apply(callback, entity);
+
+				if (value != null) {
+					return (T) value;
+				}
+
+				throw new IllegalArgumentException(
+						String.format("Callback invocation on %s returned null value for %s", callback.getClass(), entity));
+
+			} catch (ClassCastException ex) {
+
+				String msg = ex.getMessage();
+				if (msg == null || EntityCallbackInvoker.matchesClassCastMessage(msg, entity.getClass())) {
+
+					// Possibly a lambda-defined listener which we could not resolve the generic event type for
+					// -> let's suppress the exception and just log a debug message.
+					Log logger = LogFactory.getLog(getClass());
+					if (logger.isDebugEnabled()) {
+						logger.debug("Non-matching callback type for entity callback: " + callback, ex);
+					}
+					return entity;
+				} else {
+					throw ex;
+				}
+			}
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacks.java b/src/main/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacks.java
new file mode 100644
index 000000000..b5654be0e
--- /dev/null
+++ b/src/main/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacks.java
@@ -0,0 +1,141 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import reactor.core.publisher.Mono;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.core.ResolvableType;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * {@link ReactiveEntityCallbacks} implementation.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ */
+class DefaultReactiveEntityCallbacks implements ReactiveEntityCallbacks {
+
+	private final Map<Class<?>, Method> callbackMethodCache = new ConcurrentReferenceHashMap<>(64);
+	private final ReactiveEntityCallbackInvoker callbackInvoker = new DefaultReactiveEntityCallbackInvoker();
+	private final EntityCallbackDiscoverer callbackDiscoverer;
+
+	/**
+	 * Create new instance of {@link DefaultReactiveEntityCallbacks}.
+	 */
+	DefaultReactiveEntityCallbacks() {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer();
+	}
+
+	/**
+	 * Create new instance of {@link DefaultReactiveEntityCallbacks} discovering {@link EntityCallback entity callbacks}
+	 * within the given {@link BeanFactory}.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 */
+	DefaultReactiveEntityCallbacks(BeanFactory beanFactory) {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer(beanFactory);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.ReactiveEntityCallbacks#callback(java.lang.Class, java.lang.Object, java.lang.Object)
+	 */
+	@Override
+	public <T> Mono<T> callback(Class<? extends EntityCallback> callbackType, T entity, Object... args) {
+
+		Assert.notNull(entity, "Entity must not be null!");
+
+		Class<T> entityType = (Class<T>) (entity != null ? ClassUtils.getUserClass(entity.getClass())
+				: callbackDiscoverer.resolveDeclaredEntityType(callbackType).getRawClass());
+
+		Method callbackMethod = callbackMethodCache.computeIfAbsent(callbackType, it -> {
+
+			Method method = EntityCallbackDiscoverer.lookupCallbackMethod(it, entityType, args);
+			ReflectionUtils.makeAccessible(method);
+			return method;
+		});
+
+		Mono<T> deferredCallbackChain = Mono.just(entity);
+
+		for (EntityCallback<T> callback : callbackDiscoverer.getEntityCallbacks(entityType,
+				ResolvableType.forClass(callbackType))) {
+
+			BiFunction<EntityCallback<T>, T, Object> callbackFunction = EntityCallbackDiscoverer
+					.computeCallbackInvokerFunction(callback, callbackMethod, args);
+
+			deferredCallbackChain = deferredCallbackChain
+					.flatMap(it -> callbackInvoker.invokeCallback(callback, it, callbackFunction));
+		}
+
+		return deferredCallbackChain;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.EntityCallbacks#addEntityCallback(org.springframework.data.mapping.callback.EntityCallback)
+	 */
+	@Override
+	public void addEntityCallback(EntityCallback<?> callback) {
+		this.callbackDiscoverer.addEntityCallback(callback);
+	}
+
+	static class DefaultReactiveEntityCallbackInvoker implements ReactiveEntityCallbackInvoker {
+
+		@Override
+		public <T> Mono<T> invokeCallback(EntityCallback<T> callback, T entity,
+				BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction) {
+
+			try {
+
+				Object value = callbackInvokerFunction.apply(callback, entity);
+
+				if (value != null) {
+					return value instanceof Publisher ? Mono.from((Publisher<T>) value) : Mono.just((T) value);
+				}
+
+				throw new IllegalArgumentException(
+						String.format("Callback invocation on %s returned null value for %s", callback.getClass(), entity));
+			} catch (ClassCastException ex) {
+
+				String msg = ex.getMessage();
+				if (msg == null || EntityCallbackInvoker.matchesClassCastMessage(msg, entity.getClass())) {
+
+					// Possibly a lambda-defined listener which we could not resolve the generic event type for
+					// -> let's suppress the exception and just log a debug message.
+					Log logger = LogFactory.getLog(getClass());
+					if (logger.isDebugEnabled()) {
+						logger.debug("Non-matching callback type for entity callback: " + callback, ex);
+					}
+
+					return Mono.just(entity);
+				} else {
+					return Mono.error(ex);
+				}
+			}
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/mapping/callback/EntityCallback.java b/src/main/java/org/springframework/data/mapping/callback/EntityCallback.java
index f200a46a6..a6eb2e2b7 100644
--- a/src/main/java/org/springframework/data/mapping/callback/EntityCallback.java
+++ b/src/main/java/org/springframework/data/mapping/callback/EntityCallback.java
@@ -16,13 +16,37 @@
 package org.springframework.data.mapping.callback;
 
 /**
- * Marker interface for entity callbacks to be implemented in specific callback subtypes. Multiple entity callbacks are
- * invoked sequentially with the result of the previous callback.
+ * Marker interface for entity callbacks to be implemented in specific callback subtypes intended for internal usage
+ * within store specific implementations. <br />
+ * <p />
+ * Multiple entity callbacks are invoked sequentially with the result of the previous callback. Those callbacks do by
+ * default not follow an explicit order of invocation. It is strongly recommended to enforce ordering for callbacks of
+ * the same by implementing {@link org.springframework.core.Ordered} or following the annotation driven approach using
+ * {@link org.springframework.core.annotation.Order}.
+ * <p />
+ * Entity callbacks are invoked after publishing {@link org.springframework.context.ApplicationEvent events}.
+ * <p />
+ * A store specific {@link EntityCallback} needs to define a callback method accepting an object of the parameterized
+ * type as its first argument followed by additional <i>optional</i> arguments.
+ * 
+ * <pre>
+ * <code>
+ *
+ * public interface BeforeSaveCallback&lt;T&gt; extends EntityCallback&lt;T&gt; {
+ *
+ *     T onBeforeSave(T entity, String collection);
+ * }
+ * </code>
+ * </pre>
+ *
+ * The
  *
  * @author Mark Paluch
- * @param <T> Entity type.
- * @see BeforeSaveCallback
+ * @author Christoph Strobl
+ * @param <T> Entity type. Used to detect {@link EntityCallback callbacks} to invoke via their generic type signature.
  * @see org.springframework.core.Ordered
  * @see org.springframework.core.annotation.Order
  */
-public interface EntityCallback<T> {}
+public interface EntityCallback<T> {
+
+}
diff --git a/src/main/java/org/springframework/data/mapping/callback/AbstractEntityCallbacks.java b/src/main/java/org/springframework/data/mapping/callback/EntityCallbackDiscoverer.java
similarity index 78%
rename from src/main/java/org/springframework/data/mapping/callback/AbstractEntityCallbacks.java
rename to src/main/java/org/springframework/data/mapping/callback/EntityCallbackDiscoverer.java
index 387784b83..98e05a421 100644
--- a/src/main/java/org/springframework/data/mapping/callback/AbstractEntityCallbacks.java
+++ b/src/main/java/org/springframework/data/mapping/callback/EntityCallbackDiscoverer.java
@@ -15,6 +15,8 @@
  */
 package org.springframework.data.mapping.callback;
 
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedHashSet;
@@ -22,12 +24,12 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
 
 import org.springframework.aop.framework.AopProxyUtils;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
-import org.springframework.context.ApplicationContext;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotationAwareOrderComparator;
 import org.springframework.lang.Nullable;
@@ -35,73 +37,45 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ConcurrentReferenceHashMap;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.comparator.Comparators;
 
 /**
- * Abstract implementation of the for classes that wish to implement {@link EntityCallback} functionality.
- *
  * @author Mark Paluch
- * @see EntityCallback
- * @see AnnotationAwareOrderComparator
+ * @author Christoph Strobl
+ * @since 2.2
  */
-public abstract class AbstractEntityCallbacks {
-
-	@Nullable private ClassLoader beanClassLoader;
-	@Nullable private BeanFactory beanFactory;
+class EntityCallbackDiscoverer {
 
 	private final CallbackRetriever defaultRetriever = new CallbackRetriever(false);
 	private final Map<CallbackCacheKey, CallbackRetriever> retrieverCache = new ConcurrentHashMap<>(64);
 	private final Map<Class<?>, ResolvableType> entityTypeCache = new ConcurrentReferenceHashMap<>(64);
 
+	@Nullable private ClassLoader beanClassLoader;
+	@Nullable private BeanFactory beanFactory;
+
 	private Object retrievalMutex = this.defaultRetriever;
 
 	/**
 	 * Create a new {@link EntityCallback} instance.
 	 */
-	public AbstractEntityCallbacks() {}
+	EntityCallbackDiscoverer() {}
 
 	/**
-	 * Create a new {@link EntityCallback} instance given {@link ApplicationContext}. Preloads {@link EntityCallback}
-	 * beans by scanning the {@link BeanFactory}.
+	 * Create a new {@link EntityCallback} instance.
+	 * <p />
+	 * Pre loads {@link EntityCallback} beans by scanning the {@link BeanFactory}.
 	 */
-	public AbstractEntityCallbacks(ApplicationContext beanFactory) {
+	EntityCallbackDiscoverer(BeanFactory beanFactory) {
 		setBeanFactory(beanFactory);
 	}
 
-	public void setBeanClassLoader(ClassLoader classLoader) {
-		this.beanClassLoader = classLoader;
-	}
+	void addEntityCallback(EntityCallback<?> callback) {
 
-	/**
-	 * Set the {@link BeanFactory} and optionally {@link #setBeanClassLoader(ClassLoader) class loader} if not set.
-	 * Preloads {@link EntityCallback} beans by scanning the {@link BeanFactory}.
-	 *
-	 * @param beanFactory must not be {@literal null}.
-	 */
-	public void setBeanFactory(BeanFactory beanFactory) {
-		this.beanFactory = beanFactory;
-		if (beanFactory instanceof ConfigurableBeanFactory) {
-			ConfigurableBeanFactory cbf = (ConfigurableBeanFactory) beanFactory;
-			if (this.beanClassLoader == null) {
-				this.beanClassLoader = cbf.getBeanClassLoader();
-			}
-			this.retrievalMutex = cbf.getSingletonMutex();
-		}
+		Assert.notNull(callback, "Callback must not be null!");
 
-		defaultRetriever.discoverEntityCallbacks(this.beanFactory);
-		this.retrieverCache.clear();
-	}
-
-	private BeanFactory getBeanFactory() {
-		if (this.beanFactory == null) {
-			throw new IllegalStateException(
-					"EntityCallbacks cannot retrieve callback beans " + "because it is not associated with a BeanFactory");
-		}
-		return this.beanFactory;
-	}
-
-	public void addEntityCallback(EntityCallback<?> callback) {
 		synchronized (this.retrievalMutex) {
+
 			// Explicitly remove target for a proxy, if registered already,
 			// in order to avoid double invocations of the same callback.
 			Object singletonTarget = AopProxyUtils.getSingletonTarget(callback);
@@ -113,28 +87,32 @@ public abstract class AbstractEntityCallbacks {
 		}
 	}
 
-	public void addEntityCallbackBean(String callbackBeanName) {
+	void addEntityCallbackBean(String callbackBeanName) {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbackBeans.add(callbackBeanName);
 			this.retrieverCache.clear();
 		}
 	}
 
-	public void removeEntityCallback(EntityCallback<?> callback) {
+	void removeEntityCallback(EntityCallback<?> callback) {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbacks.remove(callback);
 			this.retrieverCache.clear();
 		}
 	}
 
-	public void removeEntityCallbackBean(String callbackBeanName) {
+	void removeEntityCallbackBean(String callbackBeanName) {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbackBeans.remove(callbackBeanName);
 			this.retrieverCache.clear();
 		}
 	}
 
-	public void removeAllCallbacks() {
+	void clear() {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbacks.clear();
 			this.defaultRetriever.entityCallbackBeans.clear();
@@ -142,18 +120,6 @@ public abstract class AbstractEntityCallbacks {
 		}
 	}
 
-	/**
-	 * Return a {@link Collection} of all {@link EntityCallback}s.
-	 *
-	 * @return a {@link Collection} of all {@link EntityCallback}s.
-	 * @see EntityCallback
-	 */
-	protected Collection<EntityCallback<?>> getEntityCallbacks() {
-		synchronized (this.retrievalMutex) {
-			return this.defaultRetriever.getEntityCallbacks();
-		}
-	}
-
 	/**
 	 * Return a {@link Collection} of all {@link EntityCallback}s matching the given entity type. Non-matching callbacks
 	 * get excluded early.
@@ -164,15 +130,15 @@ public abstract class AbstractEntityCallbacks {
 	 * @return a {@link Collection} of {@link EntityCallback}s.
 	 * @see EntityCallback
 	 */
-	protected Collection<EntityCallback<?>> getEntityCallbacks(Object entity, ResolvableType callbackType) {
+	<T extends S, S> Collection<EntityCallback<S>> getEntityCallbacks(Class<T> entity, ResolvableType callbackType) {
 
-		Class<?> sourceType = entity.getClass();
+		Class<?> sourceType = entity;
 		CallbackCacheKey cacheKey = new CallbackCacheKey(callbackType, sourceType);
 
 		// Quick check for existing entry on ConcurrentHashMap...
 		CallbackRetriever retriever = this.retrieverCache.get(cacheKey);
 		if (retriever != null) {
-			return retriever.getEntityCallbacks();
+			return (Collection<EntityCallback<S>>) (Collection) retriever.getEntityCallbacks();
 		}
 
 		if (this.beanClassLoader == null || (ClassUtils.isCacheSafe(entity.getClass(), this.beanClassLoader)
@@ -182,20 +148,33 @@ public abstract class AbstractEntityCallbacks {
 			synchronized (this.retrievalMutex) {
 				retriever = this.retrieverCache.get(cacheKey);
 				if (retriever != null) {
-					return retriever.getEntityCallbacks();
+					return (Collection<EntityCallback<S>>) (Collection) retriever.getEntityCallbacks();
 				}
 				retriever = new CallbackRetriever(true);
 				Collection<EntityCallback<?>> callbacks = retrieveEntityCallbacks(ResolvableType.forClass(sourceType),
 						callbackType, retriever);
 				this.retrieverCache.put(cacheKey, retriever);
-				return callbacks;
+				return (Collection<EntityCallback<S>>) (Collection) callbacks;
 			}
 		} else {
 			// No CallbackRetriever caching -> no synchronization necessary
-			return retrieveEntityCallbacks(callbackType, callbackType, null);
+			return (Collection<EntityCallback<S>>) (Collection) retrieveEntityCallbacks(callbackType, callbackType, null);
 		}
 	}
 
+	@Nullable
+	ResolvableType resolveDeclaredEntityType(Class<?> callbackType) {
+
+		ResolvableType eventType = entityTypeCache.get(callbackType);
+
+		if (eventType == null) {
+			eventType = ResolvableType.forClass(callbackType).as(EntityCallback.class).getGeneric();
+			entityTypeCache.put(callbackType, eventType);
+		}
+
+		return (eventType != ResolvableType.NONE ? eventType : null);
+	}
+
 	/**
 	 * Actually retrieve the callbacks for the given entity and callback type.
 	 *
@@ -226,7 +205,7 @@ public abstract class AbstractEntityCallbacks {
 		}
 
 		if (!callbackBeans.isEmpty()) {
-			BeanFactory beanFactory = getBeanFactory();
+			BeanFactory beanFactory = getRequiredBeanFactory();
 			for (String callbackBeanName : callbackBeans) {
 				try {
 					Class<?> callbackImplType = beanFactory.getType(callbackBeanName);
@@ -286,21 +265,136 @@ public abstract class AbstractEntityCallbacks {
 	 * @return whether the given callback should be included in the candidates for the given callback type.
 	 */
 	protected boolean supportsEvent(EntityCallback<?> callback, ResolvableType entityType, ResolvableType callbackType) {
+
 		return supportsEvent(callback.getClass(), entityType)
 				&& callbackType.isAssignableFrom(ResolvableType.forInstance(callback));
 	}
 
-	@Nullable
-	private ResolvableType resolveDeclaredEntityType(Class<?> callbackType) {
+	/**
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.beans.factory.BeanClassLoaderAware
+	 */
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = classLoader;
+	}
 
-		ResolvableType eventType = entityTypeCache.get(callbackType);
+	/**
+	 * Set the {@link BeanFactory} and optionally {@link #setBeanClassLoader(ClassLoader) class loader} if not set. Pre
+	 * loads {@link EntityCallback} beans by scanning the {@link BeanFactory}.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 * @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(BeanFactory)
+	 */
+	public void setBeanFactory(BeanFactory beanFactory) {
 
-		if (eventType == null) {
-			eventType = ResolvableType.forClass(callbackType).as(EntityCallback.class).getGeneric();
-			entityTypeCache.put(callbackType, eventType);
+		this.beanFactory = beanFactory;
+
+		if (beanFactory instanceof ConfigurableBeanFactory) {
+			ConfigurableBeanFactory cbf = (ConfigurableBeanFactory) beanFactory;
+			if (this.beanClassLoader == null) {
+				this.beanClassLoader = cbf.getBeanClassLoader();
+			}
+			this.retrievalMutex = cbf.getSingletonMutex();
 		}
 
-		return (eventType != ResolvableType.NONE ? eventType : null);
+		defaultRetriever.discoverEntityCallbacks(this.beanFactory);
+		this.retrieverCache.clear();
+	}
+
+	@Nullable
+	static Method lookupCallbackMethod(Class<?> callbackType, Class<?> entityType, Object[] args) {
+
+		Collection<Method> methods = new ArrayList<>(1);
+
+		ReflectionUtils.doWithMethods(callbackType, mc -> methods.add(mc), method -> {
+
+			if (!Modifier.isPublic(method.getModifiers()) || method.getParameterCount() != args.length + 1
+					|| method.isBridge() || ReflectionUtils.isObjectMethod(method)) {
+				return false;
+			}
+
+			return ClassUtils.isAssignable(method.getParameterTypes()[0], entityType);
+		});
+
+		if (methods.size() == 1) {
+			return methods.iterator().next();
+		}
+
+		throw new IllegalStateException(
+				String.format("%s does not define a callback method accepting %s and %s additional arguments.",
+						ClassUtils.getShortName(callbackType), ClassUtils.getShortName(entityType), args.length));
+	}
+
+	static <T> BiFunction<EntityCallback<T>, T, Object> computeCallbackInvokerFunction(EntityCallback<T> callback,
+			Method callbackMethod, Object[] args) {
+
+		return (entityCallback, entity) -> {
+
+			Object[] invocationArgs = new Object[args.length + 1];
+			invocationArgs[0] = entity;
+			if (args.length > 0) {
+				System.arraycopy(args, 0, invocationArgs, 1, args.length);
+			}
+
+			return ReflectionUtils.invokeMethod(callbackMethod, callback, invocationArgs);
+		};
+	}
+
+	private BeanFactory getRequiredBeanFactory() {
+
+		Assert.state(beanFactory != null,
+				"EntityCallbacks cannot retrieve callback beans because it is not associated with a BeanFactory");
+		return beanFactory;
+	}
+
+	/**
+	 * Helper class that encapsulates a specific set of target {@link EntityCallback callbacks}, allowing for efficient
+	 * retrieval of pre-filtered callbacks.
+	 */
+	class CallbackRetriever {
+
+		private final Set<EntityCallback<?>> entityCallbacks = new LinkedHashSet<>();
+
+		private final Set<String> entityCallbackBeans = new LinkedHashSet<>();
+
+		private final boolean preFiltered;
+
+		CallbackRetriever(boolean preFiltered) {
+			this.preFiltered = preFiltered;
+		}
+
+		Collection<EntityCallback<?>> getEntityCallbacks() {
+
+			List<EntityCallback<?>> allCallbacks = new ArrayList<>(
+					this.entityCallbacks.size() + this.entityCallbackBeans.size());
+			allCallbacks.addAll(this.entityCallbacks);
+
+			if (!this.entityCallbackBeans.isEmpty()) {
+				BeanFactory beanFactory = getRequiredBeanFactory();
+				for (String callbackBeanName : this.entityCallbackBeans) {
+					try {
+						EntityCallback<?> callback = beanFactory.getBean(callbackBeanName, EntityCallback.class);
+						if (this.preFiltered || !allCallbacks.contains(callback)) {
+							allCallbacks.add(callback);
+						}
+					} catch (NoSuchBeanDefinitionException ex) {
+						// Singleton callback instance (without backing bean definition) disappeared -
+						// probably in the middle of the destruction phase
+					}
+				}
+			}
+
+			if (!this.preFiltered || !this.entityCallbackBeans.isEmpty()) {
+				AnnotationAwareOrderComparator.sort(allCallbacks);
+			}
+
+			return allCallbacks;
+		}
+
+		void discoverEntityCallbacks(BeanFactory beanFactory) {
+			beanFactory.getBeanProvider(EntityCallback.class).stream().forEach(entityCallbacks::add);
+		}
 	}
 
 	/**
@@ -348,52 +442,4 @@ public abstract class AbstractEntityCallbacks {
 		}
 	}
 
-	/**
-	 * Helper class that encapsulates a specific set of target {@link EntityCallback callbacks}, allowing for efficient
-	 * retrieval of pre-filtered callbacks.
-	 */
-	class CallbackRetriever {
-
-		private final Set<EntityCallback<?>> entityCallbacks = new LinkedHashSet<>();
-
-		private final Set<String> entityCallbackBeans = new LinkedHashSet<>();
-
-		private final boolean preFiltered;
-
-		CallbackRetriever(boolean preFiltered) {
-			this.preFiltered = preFiltered;
-		}
-
-		Collection<EntityCallback<?>> getEntityCallbacks() {
-
-			List<EntityCallback<?>> allCallbacks = new ArrayList<>(
-					this.entityCallbacks.size() + this.entityCallbackBeans.size());
-			allCallbacks.addAll(this.entityCallbacks);
-
-			if (!this.entityCallbackBeans.isEmpty()) {
-				BeanFactory beanFactory = getBeanFactory();
-				for (String callbackBeanName : this.entityCallbackBeans) {
-					try {
-						EntityCallback<?> callback = beanFactory.getBean(callbackBeanName, EntityCallback.class);
-						if (this.preFiltered || !allCallbacks.contains(callback)) {
-							allCallbacks.add(callback);
-						}
-					} catch (NoSuchBeanDefinitionException ex) {
-						// Singleton callback instance (without backing bean definition) disappeared -
-						// probably in the middle of the destruction phase
-					}
-				}
-			}
-
-			if (!this.preFiltered || !this.entityCallbackBeans.isEmpty()) {
-				AnnotationAwareOrderComparator.sort(allCallbacks);
-			}
-
-			return allCallbacks;
-		}
-
-		void discoverEntityCallbacks(BeanFactory beanFactory) {
-			beanFactory.getBeanProvider(EntityCallback.class).stream().forEach(entityCallbacks::add);
-		}
-	}
 }
diff --git a/src/main/java/org/springframework/data/mapping/callback/EntityCallbackInvoker.java b/src/main/java/org/springframework/data/mapping/callback/EntityCallbackInvoker.java
new file mode 100644
index 000000000..5b299a74a
--- /dev/null
+++ b/src/main/java/org/springframework/data/mapping/callback/EntityCallbackInvoker.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import java.util.function.BiFunction;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+interface EntityCallbackInvoker {
+
+	/**
+	 * Invoke the actual {@link EntityCallback} for the given entity via the {@link BiFunction invoker function}.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @param entity must not be {@literal null}
+	 * @param callbackInvokerFunction must not be {@literal null}.
+	 * @param <T>
+	 * @return never {@literal null}.
+	 */
+	<T> Object invokeCallback(EntityCallback<T> callback, T entity,
+			BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction);
+
+	static boolean matchesClassCastMessage(String classCastMessage, Class<?> eventClass) {
+
+		// On Java 8, the message starts with the class name: "java.lang.String cannot be cast..."
+		if (classCastMessage.startsWith(eventClass.getName())) {
+			return true;
+		}
+
+		// On Java 11, the message starts with "class ..." a.k.a. Class.toString()
+		if (classCastMessage.startsWith(eventClass.toString())) {
+			return true;
+		}
+
+		// On Java 9, the message used to contain the module name: "java.base/java.lang.String cannot be cast..."
+		int moduleSeparatorIndex = classCastMessage.indexOf('/');
+		if (moduleSeparatorIndex != -1 && classCastMessage.startsWith(eventClass.getName(), moduleSeparatorIndex + 1)) {
+			return true;
+		}
+
+		// Assuming an unrelated class cast failure...
+		return false;
+	}
+}
diff --git a/src/main/java/org/springframework/data/mapping/callback/EntityCallbacks.java b/src/main/java/org/springframework/data/mapping/callback/EntityCallbacks.java
new file mode 100644
index 000000000..b65be1675
--- /dev/null
+++ b/src/main/java/org/springframework/data/mapping/callback/EntityCallbacks.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.util.Assert;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+public interface EntityCallbacks {
+
+	/**
+	 * Add the given {@link EntityCallback callback} using generic type argument detection for identification of supported
+	 * types.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	void addEntityCallback(EntityCallback<?> callback);
+
+	/**
+	 * Invoke matching {@link EntityCallback entity callbacks} with given arguments.
+	 *
+	 * @param callbackType must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 * @param args optional arguments.
+	 * @param <T>
+	 * @return never {@literal null}.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	<T> T callback(Class<? extends EntityCallback> callbackType, T entity, Object... args);
+
+	/**
+	 * Create a new {@link EntityCallbacks} instance with given {@link EntityCallback callbacks}. <br />
+	 * The provided {@link EntityCallback callbacks} are immediately {@link #addEntityCallback(EntityCallback) added}.
+	 */
+	static EntityCallbacks create(EntityCallback<?>... callbacks) {
+
+		EntityCallbacks entityCallbacks = create();
+		for (EntityCallback<?> callback : callbacks) {
+			entityCallbacks.addEntityCallback(callback);
+		}
+		return entityCallbacks;
+	}
+
+	/**
+	 * Obtain a new {@link EntityCallbacks} instance. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register callbacks manually.
+	 */
+	static EntityCallbacks create() {
+		return new DefaultEntityCallbacks();
+	}
+
+	/**
+	 * Obtain a new {@link EntityCallbacks} instance.
+	 * <p />
+	 * {@link EntityCallback callbacks} are pre loaded from the given {@link BeanFactory}. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register additional callbacks manually.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	static EntityCallbacks create(BeanFactory beanFactory) {
+
+		Assert.notNull(beanFactory, "Context must not be null!");
+		return new DefaultEntityCallbacks(beanFactory);
+	}
+}
diff --git a/src/main/java/org/springframework/data/mapping/callback/ReactiveBeforeSaveCallback.java b/src/main/java/org/springframework/data/mapping/callback/ReactiveBeforeSaveCallback.java
deleted file mode 100644
index fd5fa00fa..000000000
--- a/src/main/java/org/springframework/data/mapping/callback/ReactiveBeforeSaveCallback.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Copyright 2019 the original author or 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
- *
- *      https://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 org.springframework.data.mapping.callback;
-
-import org.reactivestreams.Publisher;
-
-/**
- * Entity callback before saving the entity using deferred, reactive execution. This interface allows for modifications
- * to the actual entity that are required prior to saving the entity.
- *
- * @author Mark Paluch
- * @see EntityCallback
- * @see ReactiveEntityCallbacks
- */
-@FunctionalInterface
-public interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
-
-	/**
-	 * Handle the entity callback. Modifications that result in creating a new instance should return the changed entity
-	 * as method return value.
-	 *
-	 * @param object the entity for this callback.
-	 * @return {@link Publisher} emitting the resulting entity.
-	 */
-	Publisher<T> onBeforeSave(T object);
-}
diff --git a/src/main/java/org/springframework/data/mapping/callback/BeforeSaveCallback.java b/src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbackInvoker.java
similarity index 50%
rename from src/main/java/org/springframework/data/mapping/callback/BeforeSaveCallback.java
rename to src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbackInvoker.java
index 65f66d33c..1c49591ca 100644
--- a/src/main/java/org/springframework/data/mapping/callback/BeforeSaveCallback.java
+++ b/src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbackInvoker.java
@@ -15,23 +15,28 @@
  */
 package org.springframework.data.mapping.callback;
 
+import reactor.core.publisher.Mono;
+
+import java.util.function.BiFunction;
+
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
+
 /**
- * Entity callback before saving the entity. This interface allows for modifications to the actual entity that are
- * required prior to saving the entity.
- *
- * @author Mark Paluch
- * @see EntityCallback
- * @see SimpleEntityCallbacks
+ * @author Christoph Strobl
+ * @since 2.2
  */
-@FunctionalInterface
-public interface BeforeSaveCallback<T> extends EntityCallback<T> {
+interface ReactiveEntityCallbackInvoker extends EntityCallbackInvoker {
 
 	/**
-	 * Handle the entity callback. Modifications that result in creating a new instance should return the changed entity
-	 * as method return value.
-	 *
-	 * @param object the entity for this callback.
-	 * @return the resulting entity.
+	 * @param callback must not be {@literal null}.
+	 * @param entity can be {@literal null}
+	 * @param callbackInvokerFunction must not be {@literal null}.
+	 * @param <T>
+	 * @return a {@link Mono} emitting the result of the invocation.
 	 */
-	T onBeforeSave(T object);
+	@NonNull
+	@Override
+	<T> Mono<T> invokeCallback(EntityCallback<T> callback, @Nullable T entity,
+			BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction);
 }
diff --git a/src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacks.java b/src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacks.java
index 593632e36..7626844ee 100644
--- a/src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacks.java
+++ b/src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacks.java
@@ -17,60 +17,70 @@ package org.springframework.data.mapping.callback;
 
 import reactor.core.publisher.Mono;
 
-import java.util.function.BiFunction;
-
-import org.reactivestreams.Publisher;
-import org.springframework.context.ApplicationContext;
-import org.springframework.core.ResolvableType;
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.util.Assert;
 
 /**
- * Reactive {@link SimpleEntityCallbacks} implementation.
- *
- * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.2
  */
-public class ReactiveEntityCallbacks extends SimpleEntityCallbacks {
+public interface ReactiveEntityCallbacks {
 
-	public ReactiveEntityCallbacks() {
-		super();
-	}
+	/**
+	 * Add the given {@link EntityCallback callback} using generic type argument detection for identification of supported
+	 * types.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	void addEntityCallback(EntityCallback<?> callback);
 
-	public ReactiveEntityCallbacks(ApplicationContext beanFactory) {
-		super(beanFactory);
+	/**
+	 * On {@link Mono#subscribe() subscribe} invoke the matching {@link EntityCallback entity callbacks} with given
+	 * arguments.
+	 *
+	 * @param callbackType must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 * @param args optional arguments.
+	 * @param <T>
+	 * @return a {@link Mono} emitting the result after invoking the callbacks.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	<T> Mono<T> callback(Class<? extends EntityCallback> callbackType, T entity, Object... args);
+
+	/**
+	 * Create a new {@link ReactiveEntityCallbacks} instance with given {@link EntityCallback callbacks}. <br />
+	 * The provided {@link EntityCallback callbacks} are immediately {@link #addEntityCallback(EntityCallback) added}.
+	 */
+	static ReactiveEntityCallbacks create(EntityCallback<?>... callbacks) {
+
+		ReactiveEntityCallbacks entityCallbacks = create();
+		for (EntityCallback<?> callback : callbacks) {
+			entityCallbacks.addEntityCallback(callback);
+		}
+		return entityCallbacks;
 	}
 
 	/**
-	 * Perform an entity callback.
-	 *
-	 * @param entity the entity, must not be {@literal null}.
-	 * @param callbackType desired callback type.
-	 * @param callbackInvoker invocation function for the callback to optionally pass additional parameters.
-	 * @return the resulting entity after invoking all callbacks.
+	 * Obtain a new {@link ReactiveEntityCallbacks} instance. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register callbacks manually.
 	 */
-	@SuppressWarnings("unchecked")
-	public <T, E extends EntityCallback<T>> Mono<T> callbackLater(T entity, Class<? extends E> callbackType,
-			BiFunction<? extends E, T, Publisher<? extends Object>> callbackInvoker) {
+	static ReactiveEntityCallbacks create() {
+		return new DefaultReactiveEntityCallbacks();
+	}
 
-		Assert.notNull(entity, "Entity must not be null!");
+	/**
+	 * Obtain a new {@link ReactiveEntityCallbacks} instance.
+	 * <p />
+	 * {@link EntityCallback callbacks} are pre loaded from the given {@link BeanFactory}. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register additional callbacks manually.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	static ReactiveEntityCallbacks create(BeanFactory beanFactory) {
 
-		ResolvableType resolvedCallbackType = ResolvableType.forClass(callbackType);
-		Mono<T> deferredCallbackChain = Mono.just(entity);
-
-		for (EntityCallback<?> callback : getEntityCallbacks(entity, resolvedCallbackType)) {
-
-			deferredCallbackChain = deferredCallbackChain.flatMap(it -> {
-
-				Object o = invokeCallback(callback, it, (BiFunction) callbackInvoker);
-
-				if (o instanceof Publisher) {
-					return Mono.from((Publisher<T>) o);
-				}
-
-				throw new IllegalStateException("Callback " + callback + " returned a non-Publisher type " + o);
-
-			});
-		}
-
-		return deferredCallbackChain;
+		Assert.notNull(beanFactory, "Context must not be null!");
+		return new DefaultReactiveEntityCallbacks(beanFactory);
 	}
 }
diff --git a/src/main/java/org/springframework/data/mapping/callback/SimpleEntityCallbacks.java b/src/main/java/org/springframework/data/mapping/callback/SimpleEntityCallbacks.java
deleted file mode 100644
index a26ef9aca..000000000
--- a/src/main/java/org/springframework/data/mapping/callback/SimpleEntityCallbacks.java
+++ /dev/null
@@ -1,159 +0,0 @@
-/*
- * Copyright 2019 the original author or 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
- *
- *      https://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 org.springframework.data.mapping.callback;
-
-import java.util.function.BiFunction;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.context.ApplicationContext;
-import org.springframework.core.ResolvableType;
-import org.springframework.lang.Nullable;
-import org.springframework.util.Assert;
-import org.springframework.util.ErrorHandler;
-
-/**
- * Simple implementation to invoke {@link EntityCallback}s.
- *
- * @author Mark Paluch
- */
-public class SimpleEntityCallbacks extends AbstractEntityCallbacks {
-
-	@Nullable private ErrorHandler errorHandler;
-
-	public SimpleEntityCallbacks() {
-		super();
-	}
-
-	public SimpleEntityCallbacks(ApplicationContext beanFactory) {
-		super(beanFactory);
-	}
-
-	/**
-	 * Set the {@link ErrorHandler} to invoke in case an exception is thrown from a {@link EntityCallback}.
-	 * <p>
-	 * Default is none, with a callback {@link Exception} stopping the current dispatch and getting propagated to the
-	 * publisher of the current event.
-	 * <p>
-	 * Consider setting an {@link ErrorHandler} implementation that catches and logs exceptions (a la
-	 * {@link org.springframework.scheduling.support.TaskUtils#LOG_AND_SUPPRESS_ERROR_HANDLER}) or an implementation that
-	 * logs exceptions while nevertheless propagating them (e.g.
-	 * {@link org.springframework.scheduling.support.TaskUtils#LOG_AND_PROPAGATE_ERROR_HANDLER}).
-	 */
-	public void setErrorHandler(@Nullable ErrorHandler errorHandler) {
-		this.errorHandler = errorHandler;
-	}
-
-	/**
-	 * Return the current {@link ErrorHandler} for this callback dispatcher.
-	 *
-	 * @return the current {@link ErrorHandler}.
-	 */
-	@Nullable
-	protected ErrorHandler getErrorHandler() {
-		return this.errorHandler;
-	}
-
-	/**
-	 * Perform an entity callback.
-	 *
-	 * @param entity the entity, must not be {@literal null}.
-	 * @param callbackType desired callback type.
-	 * @param callbackInvoker invocation function for the callback to optionally pass additional parameters.
-	 * @return the resulting entity after invoking all callbacks.
-	 */
-	@SuppressWarnings("unchecked")
-	public <T, E extends EntityCallback<T>> T callback(T entity, Class<? extends E> callbackType,
-			BiFunction<? extends E, T, Object> callbackInvoker) {
-
-		Assert.notNull(entity, "Entity must not be null!");
-
-		ResolvableType resolvedCallbackType = ResolvableType.forClass(callbackType);
-		T entityToUse = entity;
-
-		for (EntityCallback<?> callback : getEntityCallbacks(entity, resolvedCallbackType)) {
-			entityToUse = (T) invokeCallback(callback, entityToUse, (BiFunction) callbackInvoker);
-		}
-
-		return entityToUse;
-	}
-
-	/**
-	 * Invoke the given callback with the given entity.
-	 */
-	protected Object invokeCallback(EntityCallback<?> callback, Object entity,
-			BiFunction<EntityCallback<?>, Object, Object> callbackInvokerFunction) {
-
-		ErrorHandler errorHandler = getErrorHandler();
-
-		if (errorHandler != null) {
-			try {
-				return doInvokeCallback(callback, entity, callbackInvokerFunction);
-			} catch (Throwable err) {
-				errorHandler.handleError(err);
-				return entity;
-			}
-		}
-
-		return doInvokeCallback(callback, entity, callbackInvokerFunction);
-
-	}
-
-	@SuppressWarnings({ "rawtypes" })
-	private Object doInvokeCallback(EntityCallback<?> callback, Object entity,
-			BiFunction<EntityCallback<?>, Object, Object> callbackInvokerFunction) {
-
-		try {
-			return callbackInvokerFunction.apply(callback, entity);
-		} catch (ClassCastException ex) {
-			String msg = ex.getMessage();
-			if (msg == null || matchesClassCastMessage(msg, entity.getClass())) {
-				// Possibly a lambda-defined listener which we could not resolve the generic event type for
-				// -> let's suppress the exception and just log a debug message.
-				Log logger = LogFactory.getLog(getClass());
-				if (logger.isDebugEnabled()) {
-					logger.debug("Non-matching callback type for entity callback: " + callback, ex);
-				}
-
-				return entity;
-			} else {
-				throw ex;
-			}
-		}
-	}
-
-	private boolean matchesClassCastMessage(String classCastMessage, Class<?> eventClass) {
-
-		// On Java 8, the message starts with the class name: "java.lang.String cannot be cast..."
-		if (classCastMessage.startsWith(eventClass.getName())) {
-			return true;
-		}
-
-		// On Java 11, the message starts with "class ..." a.k.a. Class.toString()
-		if (classCastMessage.startsWith(eventClass.toString())) {
-			return true;
-		}
-
-		// On Java 9, the message used to contain the module name: "java.base/java.lang.String cannot be cast..."
-		int moduleSeparatorIndex = classCastMessage.indexOf('/');
-		if (moduleSeparatorIndex != -1 && classCastMessage.startsWith(eventClass.getName(), moduleSeparatorIndex + 1)) {
-			return true;
-		}
-
-		// Assuming an unrelated class cast failure...
-		return false;
-	}
-}
diff --git a/src/main/java/org/springframework/data/mapping/callback/package-info.java b/src/main/java/org/springframework/data/mapping/callback/package-info.java
new file mode 100644
index 000000000..ade249bfc
--- /dev/null
+++ b/src/main/java/org/springframework/data/mapping/callback/package-info.java
@@ -0,0 +1,5 @@
+/**
+ * Mapping callback API and implementation base classes.
+ */
+@org.springframework.lang.NonNullApi
+package org.springframework.data.mapping.callback;
diff --git a/src/test/java/org/springframework/data/mapping/callback/CapturingEntityCallback.java b/src/test/java/org/springframework/data/mapping/callback/CapturingEntityCallback.java
new file mode 100644
index 000000000..226012d77
--- /dev/null
+++ b/src/test/java/org/springframework/data/mapping/callback/CapturingEntityCallback.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.core.Ordered;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * @author Christoph Strobl
+ */
+class CapturingEntityCallback implements EntityCallback<Person> {
+
+	final List<Person> captured = new ArrayList<>(3);
+	final @Nullable Person returnValue;
+
+	CapturingEntityCallback() {
+		this(new PersonDocument(null, null, null));
+	}
+
+	CapturingEntityCallback(@Nullable Person returnValue) {
+		this.returnValue = returnValue;
+	}
+
+	public Person doSomething(Person person) {
+
+		captured.add(person);
+		return returnValue;
+	}
+
+	Person capturedValue() {
+		return CollectionUtils.lastElement(captured);
+	}
+
+	List<Person> capturedValues() {
+		return captured;
+	}
+
+	static class FirstCallback extends CapturingEntityCallback implements Ordered {
+
+		@Override
+		public int getOrder() {
+			return 1;
+		}
+	}
+
+	static class SecondCallback extends CapturingEntityCallback implements Ordered {
+
+		public SecondCallback() {}
+
+		public SecondCallback(Person returnValue) {
+			super(returnValue);
+		}
+
+		@Override
+		public int getOrder() {
+			return 2;
+		}
+	}
+
+	static class ThirdCallback extends CapturingEntityCallback implements Ordered {
+
+		@Override
+		public int getOrder() {
+			return 3;
+		}
+	}
+}
diff --git a/src/test/java/org/springframework/data/mapping/callback/DefaultEntityCallbacksUnitTests.java b/src/test/java/org/springframework/data/mapping/callback/DefaultEntityCallbacksUnitTests.java
new file mode 100644
index 000000000..b9f977163
--- /dev/null
+++ b/src/test/java/org/springframework/data/mapping/callback/DefaultEntityCallbacksUnitTests.java
@@ -0,0 +1,291 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.Ordered;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.FirstCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.SecondCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.ThirdCallback;
+
+/**
+ * Unit tests for {@link DefaultEntityCallbacks}.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ */
+public class DefaultEntityCallbacksUnitTests {
+
+	@Test // DATACMNS-1467
+	public void shouldDispatchCallback() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		PersonDocument afterCallback = callbacks.callback(BeforeSaveCallback.class, personDocument);
+
+		assertThat(afterCallback.getSsn()).isEqualTo(6);
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldDispatchCallsToLambdaReceivers() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(LambdaConfig.class);
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		PersonDocument afterCallback = callbacks.callback(BeforeSaveCallback.class, personDocument);
+
+		assertThat(afterCallback).isSameAs(personDocument);
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeGenericEvent() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new GenericPersonCallback());
+
+		Person afterCallback = callbacks.callback(GenericPersonCallback.class, new PersonDocument(null, "Walter", null));
+
+		assertThat(afterCallback.getSsn()).isEqualTo(6);
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeGenericEventWithArgs() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new GenericPersonCallbackWithArgs());
+
+		Person afterCallback = callbacks.callback(GenericPersonCallbackWithArgs.class,
+				new PersonDocument(null, "Walter", null), "agr0", Float.POSITIVE_INFINITY);
+
+		assertThat(afterCallback.getSsn()).isEqualTo(6);
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeInvalidEvent() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new InvalidEntityCallback() {});
+
+		assertThatExceptionOfType(IllegalStateException.class)
+				.isThrownBy(() -> callbacks.callback(InvalidEntityCallback.class, new PersonDocument(null, "Walter", null),
+						"agr0", Float.POSITIVE_INFINITY));
+	}
+
+	@Test // DATACMNS-1467
+	public void passesInvocationResultOnAlongTheChain() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback();
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		callbacks.callback(CapturingEntityCallback.class, initial);
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(first.capturedValues()).hasSize(1);
+		assertThat(second.capturedValue()).isNotSameAs(initial);
+		assertThat(second.capturedValues()).hasSize(1);
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullEntity() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new CapturingEntityCallback());
+
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> callbacks.callback(CapturingEntityCallback.class, null));
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullValueReturnedByCallbackEntity() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback(null);
+		CapturingEntityCallback third = new ThirdCallback();
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+		callbacks.addEntityCallback(third);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> callbacks.callback(CapturingEntityCallback.class, initial));
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(second.capturedValue()).isNotNull().isNotSameAs(initial);
+		assertThat(third.capturedValues()).isEmpty();
+	}
+
+	@Test // DATACMNS-1467
+	public void detectsMultipleCallbacksWithinOneClass() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MultipleCallbacksInOneClassConfig.class);
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		callbacks.callback(BeforeSaveCallback.class, personDocument);
+
+		assertThat(ctx.getBean("callbacks", MultipleCallbacks.class).invocations).containsExactly("save");
+
+		callbacks.callback(BeforeConvertCallback.class, personDocument);
+
+		assertThat(ctx.getBean("callbacks", MultipleCallbacks.class).invocations).containsExactly("save", "convert");
+	}
+
+	@Configuration
+	static class MyConfig {
+
+		@Bean
+		MyBeforeSaveCallback callback() {
+			return new MyBeforeSaveCallback();
+		}
+
+		@Bean
+		@Lazy
+		Object namedCallback() {
+			return new MyOtherCallback();
+		}
+	}
+
+	@Configuration
+	static class LambdaConfig {
+
+		@Bean
+		BeforeSaveCallback<User> userCallback() {
+			return object -> object;
+		}
+
+		@Bean
+		BeforeSaveCallback<Person> personCallback() {
+
+			return object -> {
+				object.setSsn(object.getFirstName().length());
+				return object;
+			};
+		}
+	}
+
+	@Configuration
+	static class MultipleCallbacksInOneClassConfig {
+
+		@Bean
+		MultipleCallbacks callbacks() {
+			return new MultipleCallbacks();
+		}
+	}
+
+	interface BeforeConvertCallback<T> extends EntityCallback<T>, Ordered {
+		T onBeforeConvert(T object);
+
+		@Override
+		default int getOrder() {
+			return 0;
+		}
+	}
+
+	interface BeforeSaveCallback<T> extends EntityCallback<T> {
+		T onBeforeSave(T object);
+	}
+
+	static class MyBeforeSaveCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+
+			object.setSsn(object.getFirstName().length());
+			return object;
+		}
+	}
+
+	static class MyOtherCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+			return object;
+		}
+	}
+
+	static class User {}
+
+	static class GenericPersonCallback implements EntityCallback<Person> {
+
+		public Person onBeforeSave(Person value) {
+
+			value.setSsn(value.getFirstName().length());
+			return value;
+		}
+	}
+
+	static class GenericPersonCallbackWithArgs implements EntityCallback<Person> {
+
+		public Person onBeforeSave(Person value, String agr1, Object arg2) {
+
+			value.setSsn(value.getFirstName().length());
+			return value;
+		}
+	}
+
+	interface InvalidEntityCallback extends EntityCallback<Person> {
+
+		default Person onBeforeSave(String value, Person entity) {
+			return entity;
+		}
+	}
+
+	static class MultipleCallbacks implements BeforeConvertCallback<Person>, BeforeSaveCallback<Person> {
+
+		List<String> invocations = new ArrayList(2);
+
+		@Override
+		public Person onBeforeConvert(Person object) {
+
+			invocations.add("convert");
+			return object;
+		}
+
+		@Override
+		public Person onBeforeSave(Person object) {
+
+			invocations.add("save");
+			return object;
+		}
+	}
+
+}
diff --git a/src/test/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacksUnitTests.java b/src/test/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacksUnitTests.java
new file mode 100644
index 000000000..644de4ec1
--- /dev/null
+++ b/src/test/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacksUnitTests.java
@@ -0,0 +1,161 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import static org.assertj.core.api.Assertions.*;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.junit.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.FirstCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.SecondCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.ThirdCallback;
+
+/**
+ * Unit tests for {@link DefaultReactiveEntityCallbacks}.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ */
+public class DefaultReactiveEntityCallbacksUnitTests {
+
+	@Test // DATACMNS-1467
+	public void dispatchResolvesOnSubscribe() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		Mono<PersonDocument> afterCallback = callbacks.callback(ReactiveBeforeSaveCallback.class, personDocument);
+
+		assertThat(personDocument.getSsn()).isNull();
+
+		afterCallback.as(StepVerifier::create) //
+				.consumeNextWith(it -> assertThat(it.getSsn()).isEqualTo(6)) //
+				.verifyComplete();
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeGenericEvent() {
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(new GenericPersonCallback());
+
+		callbacks.callback(GenericPersonCallback.class, new PersonDocument(null, "Walter", null)) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> assertThat(it.getSsn()).isEqualTo(6)) //
+				.verifyComplete();
+	}
+
+	@Test // DATACMNS-1467
+	public void passesInvocationResultOnAlongTheChain() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback();
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		callbacks.callback(CapturingEntityCallback.class, initial) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(first.capturedValues()).hasSize(1);
+		assertThat(second.capturedValue()).isNotSameAs(initial);
+		assertThat(second.capturedValues()).hasSize(1);
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullEntity() {
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(new CapturingEntityCallback());
+
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> callbacks.callback(CapturingEntityCallback.class, null));
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullValueReturnedByCallbackEntity() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback(null);
+		CapturingEntityCallback third = new ThirdCallback();
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+		callbacks.addEntityCallback(third);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		callbacks.callback(CapturingEntityCallback.class, initial) //
+				.as(StepVerifier::create) //
+				.expectError(IllegalArgumentException.class) //
+				.verify();
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(second.capturedValue()).isNotNull().isNotSameAs(initial);
+		assertThat(third.capturedValues()).isEmpty();
+	}
+
+	@Configuration
+	static class MyConfig {
+
+		@Bean
+		MyReactiveBeforeSaveCallback callback() {
+			return new MyReactiveBeforeSaveCallback();
+		}
+
+	}
+
+	interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
+		Mono<T> onBeforeSave(T object);
+	}
+
+	static class MyReactiveBeforeSaveCallback implements ReactiveBeforeSaveCallback<Person> {
+
+		@Override
+		public Mono<Person> onBeforeSave(Person object) {
+
+			PersonDocument result = new PersonDocument(object.getFirstName().length(), object.getFirstName(),
+					object.getLastName());
+
+			return Mono.just(result);
+		}
+	}
+
+	static class GenericPersonCallback implements EntityCallback<Person> {
+
+		public Person onBeforeSave(Person value) {
+
+			value.setSsn(value.getFirstName().length());
+			return value;
+		}
+	}
+}
diff --git a/src/test/java/org/springframework/data/mapping/callback/EntityCallbackDiscovererUnitTests.java b/src/test/java/org/springframework/data/mapping/callback/EntityCallbackDiscovererUnitTests.java
new file mode 100644
index 000000000..eead4aa04
--- /dev/null
+++ b/src/test/java/org/springframework/data/mapping/callback/EntityCallbackDiscovererUnitTests.java
@@ -0,0 +1,207 @@
+/*
+ * Copyright 2019 the original author or 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
+ *
+ *      https://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 org.springframework.data.mapping.callback;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.Collection;
+
+import org.junit.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.Ordered;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.annotation.Order;
+import org.springframework.data.mapping.Child;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+
+/**
+ * @author Christoph Strobl
+ */
+public class EntityCallbackDiscovererUnitTests {
+
+	@Test // DATACMNS-1467
+	public void shouldDiscoverCallbackType() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer(ctx);
+
+		Collection<EntityCallback<Person>> entityCallbacks = discoverer.getEntityCallbacks(PersonDocument.class,
+				ResolvableType.forType(BeforeSaveCallback.class));
+
+		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyBeforeSaveCallback.class);
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldDiscoverCallbackTypeByName() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer(ctx);
+		discoverer.clear();
+		discoverer.addEntityCallbackBean("namedCallback");
+
+		Collection<EntityCallback<Person>> entityCallbacks = discoverer.getEntityCallbacks(PersonDocument.class,
+				ResolvableType.forType(BeforeSaveCallback.class));
+
+		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyOtherCallback.class);
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldSupportCallbackTypes() {
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer();
+
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Person.class))).isTrue();
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Child.class))).isTrue();
+		assertThat(discoverer.supportsEvent(BeforeSaveCallback.class, ResolvableType.forClass(PersonDocument.class)))
+				.isTrue();
+
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Object.class))).isFalse();
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(User.class))).isFalse();
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldSupportInstanceCallbackTypes() {
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer();
+
+		MyBeforeSaveCallback callback = new MyBeforeSaveCallback();
+
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(Person.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(Child.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(User.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isFalse();
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldDispatchInOrder() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(OrderedConfig.class);
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer(ctx);
+
+		Collection<EntityCallback<Person>> entityCallbacks = discoverer.getEntityCallbacks(PersonDocument.class,
+				ResolvableType.forType(EntityCallback.class));
+
+		assertThat(entityCallbacks).containsExactly(ctx.getBean("callback1", EntityCallback.class),
+				ctx.getBean("callback2", EntityCallback.class), ctx.getBean("callback3", EntityCallback.class),
+				ctx.getBean("callback4", EntityCallback.class));
+	}
+
+	@Configuration
+	static class MyConfig {
+
+		@Bean
+		MyBeforeSaveCallback callback() {
+			return new MyBeforeSaveCallback();
+		}
+
+		@Bean
+		@Lazy
+		Object namedCallback() {
+			return new MyOtherCallback();
+		}
+	}
+
+	@Configuration
+	static class OrderedConfig {
+
+		@Bean
+		EntityCallback<Person> callback4() {
+
+			return (BeforeSaveCallback<Person>) object -> object;
+		}
+
+		@Bean
+		EntityCallback<Person> callback2() {
+			return new Second();
+		}
+
+		@Bean
+		EntityCallback<Person> callback3() {
+			return new Third();
+		}
+
+		@Bean
+		EntityCallback<Person> callback1() {
+			return new First();
+		}
+
+		@Order(3)
+		static class Third implements EntityCallback<Person> {
+
+			public Person beforeSave(Person object) {
+				return object;
+			}
+		}
+
+		static class Second implements EntityCallback<Person>, Ordered {
+
+			public Person beforeSave(Person object) {
+				return object;
+			}
+
+			@Override
+			public int getOrder() {
+				return 2;
+			}
+		}
+
+		@Order(1)
+		static class First implements EntityCallback<Person> {
+
+			public Person beforeSave(Person object) {
+				return object;
+			}
+		}
+	}
+
+	interface BeforeSaveCallback<T> extends EntityCallback<T> {
+		T onBeforeSave(T object);
+	}
+
+	static class MyBeforeSaveCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+
+			object.setSsn(object.getFirstName().length());
+			return object;
+		}
+	}
+
+	static class MyOtherCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+			return object;
+		}
+	}
+
+	static class User {}
+}
diff --git a/src/test/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacksUnitTests.java b/src/test/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacksUnitTests.java
deleted file mode 100644
index f45156b99..000000000
--- a/src/test/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacksUnitTests.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Copyright 2019 the original author or 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
- *
- *      https://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 org.springframework.data.mapping.callback;
-
-import static org.assertj.core.api.Assertions.*;
-
-import reactor.core.publisher.Mono;
-
-import org.junit.Test;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.data.mapping.Person;
-import org.springframework.data.mapping.PersonDocument;
-
-/**
- * Unit tests for {@link ReactiveEntityCallbacks}.
- *
- * @author Mark Paluch
- */
-public class ReactiveEntityCallbacksUnitTests {
-
-	@Test
-	public void shouldDispatchCallback() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		ReactiveEntityCallbacks callbacks = new ReactiveEntityCallbacks(ctx);
-
-		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
-		Mono<PersonDocument> afterCallback = callbacks.callbackLater(personDocument, ReactiveBeforeSaveCallback.class,
-				ReactiveBeforeSaveCallback::onBeforeSave);
-
-		assertThat(personDocument.getSsn()).isNull();
-		assertThat(afterCallback.block().getSsn()).isEqualTo(6);
-	}
-
-	@Configuration
-	static class MyConfig {
-
-		@Bean
-		MyReactiveBeforeSaveCallback callback() {
-			return new MyReactiveBeforeSaveCallback();
-		}
-
-	}
-
-	static class MyReactiveBeforeSaveCallback implements ReactiveBeforeSaveCallback<Person> {
-
-		@Override
-		public Mono<Person> onBeforeSave(Person object) {
-
-			PersonDocument result = new PersonDocument(object.getFirstName().length(), object.getFirstName(),
-					object.getLastName());
-
-			return Mono.just(result);
-		}
-	}
-}
diff --git a/src/test/java/org/springframework/data/mapping/callback/SimpleEntityCallbacksUnitTests.java b/src/test/java/org/springframework/data/mapping/callback/SimpleEntityCallbacksUnitTests.java
deleted file mode 100644
index 8d8cb5fd6..000000000
--- a/src/test/java/org/springframework/data/mapping/callback/SimpleEntityCallbacksUnitTests.java
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- * Copyright 2019 the original author or 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
- *
- *      https://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 org.springframework.data.mapping.callback;
-
-import static org.assertj.core.api.Assertions.*;
-
-import java.util.Collection;
-
-import org.junit.Test;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Lazy;
-import org.springframework.core.ResolvableType;
-import org.springframework.data.mapping.Child;
-import org.springframework.data.mapping.Person;
-import org.springframework.data.mapping.PersonDocument;
-
-/**
- * Unit tests for {@link SimpleEntityCallbacks}.
- *
- * @author Mark Paluch
- */
-public class SimpleEntityCallbacksUnitTests {
-
-	@Test
-	public void shouldDispatchCallback() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-
-		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
-		PersonDocument afterCallback = callbacks.callback(personDocument, BeforeSaveCallback.class,
-				BeforeSaveCallback::onBeforeSave);
-
-		assertThat(afterCallback.getSsn()).isEqualTo(6);
-	}
-
-	@Test
-	public void shouldDispatchCallsToLambdaReceivers() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(LambdaConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-
-		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
-		PersonDocument afterCallback = callbacks.callback(personDocument, BeforeSaveCallback.class,
-				BeforeSaveCallback::onBeforeSave);
-
-		assertThat(afterCallback.getSsn()).isEqualTo(6);
-	}
-
-	@Test
-	public void shouldDiscoverCallbackType() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-
-		Collection<EntityCallback<?>> entityCallbacks = callbacks.getEntityCallbacks(new PersonDocument(null, null, null),
-				ResolvableType.forType(BeforeSaveCallback.class));
-
-		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyBeforeSaveCallback.class);
-	}
-
-	@Test
-	public void shouldDiscoverCallbackTypeByName() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-		callbacks.removeAllCallbacks();
-		callbacks.addEntityCallbackBean("namedCallback");
-
-		Collection<EntityCallback<?>> entityCallbacks = callbacks.getEntityCallbacks(new PersonDocument(null, null, null),
-				ResolvableType.forType(BeforeSaveCallback.class));
-
-		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyOtherCallback.class);
-	}
-
-	@Test
-	public void shouldSupportCallbackTypes() {
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks();
-
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Person.class))).isTrue();
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Child.class))).isTrue();
-		assertThat(callbacks.supportsEvent(BeforeSaveCallback.class, ResolvableType.forClass(PersonDocument.class)))
-				.isTrue();
-
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Object.class))).isFalse();
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(User.class))).isFalse();
-	}
-
-	@Test
-	public void shouldSupportInstanceCallbackTypes() {
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks();
-
-		MyBeforeSaveCallback callback = new MyBeforeSaveCallback();
-
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(Person.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(Child.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(User.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isFalse();
-	}
-
-	@Configuration
-	static class MyConfig {
-
-		@Bean
-		MyBeforeSaveCallback callback() {
-			return new MyBeforeSaveCallback();
-		}
-
-		@Bean
-		@Lazy
-		Object namedCallback() {
-			return new MyOtherCallback();
-		}
-	}
-
-	@Configuration
-	static class LambdaConfig {
-
-		@Bean
-		BeforeSaveCallback<User> userCallback() {
-			return object -> object;
-		}
-
-		@Bean
-		BeforeSaveCallback<Person> personCallback() {
-			return object -> {
-				object.setSsn(object.getFirstName().length());
-				return object;
-			};
-		}
-	}
-
-	static class MyBeforeSaveCallback implements BeforeSaveCallback<Person> {
-
-		@Override
-		public Person onBeforeSave(Person object) {
-			object.setSsn(object.getFirstName().length());
-			return object;
-		}
-	}
-
-	static class MyOtherCallback implements BeforeSaveCallback<Person> {
-
-		@Override
-		public Person onBeforeSave(Person object) {
-			return object;
-		}
-	}
-
-	static class User {}
-}