DATAMONGO-1928 - Use non-blocking index creation through ReactiveMongoTemplate.

We now use ReactiveIndexOperationsProvider to inspect and create indexes for MongoDB collections without using blocking methods. Indexes are created for initial entities and whenever a MongoPersistentEntity is registered in MongoMappingContext.

Index creation is now decoupled from the actual ReactiveMongoTemplate call causing indexes to be created asynchronously. Mongo commands no longer depend on the completion of index creation commands. Decoupling also comes with the aspect that ReactiveMongoTemplate creation/command invocation no longer fails if the actual index creation fails. Previous usage of blocking index creation caused the actual ReactiveMongoTemplate call to fail.

ReactiveMongoTemplate objects can be created with a Consumer<Throwable> callback that is notified if an index creation fails.

Original Pull Request: #550

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -39,6 +39,7 @@ import org.bson.codecs.Codec;
 import org.bson.conversions.Bson;
 import org.bson.types.ObjectId;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
@@ -59,10 +60,12 @@ import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResult;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.mapping.MappingException;
+import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.PropertyPath;
 import org.springframework.data.mapping.PropertyReferenceException;
 import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mapping.context.MappingContextEvent;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
@@ -73,10 +76,9 @@ import org.springframework.data.mongodb.core.aggregation.PrefixingDelegatingAggr
 import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.*;
-import org.springframework.data.mongodb.core.index.IndexOperationsAdapter;
 import org.springframework.data.mongodb.core.index.MongoMappingEventPublisher;
-import org.springframework.data.mongodb.core.index.MongoPersistentEntityIndexCreator;
 import org.springframework.data.mongodb.core.index.ReactiveIndexOperations;
+import org.springframework.data.mongodb.core.index.ReactiveMongoPersistentEntityIndexCreator;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
@@ -185,13 +187,14 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	private final UpdateMapper updateMapper;
 	private final JsonSchemaMapper schemaMapper;
 	private final SpelAwareProxyProjectionFactory projectionFactory;
+	private final ApplicationListener<MappingContextEvent<?, ?>> indexCreatorListener;
 
 	private @Nullable WriteConcern writeConcern;
 	private WriteConcernResolver writeConcernResolver = DefaultWriteConcernResolver.INSTANCE;
 	private WriteResultChecking writeResultChecking = WriteResultChecking.NONE;
 	private @Nullable ReadPreference readPreference;
 	private @Nullable ApplicationEventPublisher eventPublisher;
-	private @Nullable MongoPersistentEntityIndexCreator indexCreator;
+	private @Nullable ReactiveMongoPersistentEntityIndexCreator indexCreator;
 
 	/**
 	 * Constructor used for a basic template configuration.
@@ -220,6 +223,21 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	 */
 	public ReactiveMongoTemplate(ReactiveMongoDatabaseFactory mongoDatabaseFactory,
 			@Nullable MongoConverter mongoConverter) {
+		this(mongoDatabaseFactory, mongoConverter, ReactiveMongoTemplate::handleSubscriptionException);
+	}
+
+	/**
+	 * Constructor used for a basic template configuration.
+	 *
+	 * @param mongoDatabaseFactory must not be {@literal null}.
+	 * @param mongoConverter can be {@literal null}.
+	 * @param subscriptionExceptionHandler exception handler called by {@link Flux#doOnError(Consumer)} on reactive type
+	 *          materialization via {@link Publisher#subscribe(Subscriber)}. This callback is used during non-blocking
+	 *          subscription of e.g. index creation {@link Publisher}s. Must not be {@literal null}.
+	 * @since 2.1
+	 */
+	public ReactiveMongoTemplate(ReactiveMongoDatabaseFactory mongoDatabaseFactory,
+			@Nullable MongoConverter mongoConverter, Consumer<Throwable> subscriptionExceptionHandler) {
 
 		Assert.notNull(mongoDatabaseFactory, "ReactiveMongoDatabaseFactory must not be null!");
 
@@ -230,18 +248,21 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		this.updateMapper = new UpdateMapper(this.mongoConverter);
 		this.schemaMapper = new MongoJsonSchemaMapper(this.mongoConverter);
 		this.projectionFactory = new SpelAwareProxyProjectionFactory();
+		this.indexCreatorListener = new IndexCreatorEventListener(subscriptionExceptionHandler);
 
 		// We always have a mapping context in the converter, whether it's a simple one or not
-		mappingContext = this.mongoConverter.getMappingContext();
-		// We create indexes based on mapping events
+		this.mappingContext = this.mongoConverter.getMappingContext();
 
-		if (mappingContext instanceof MongoMappingContext) {
-			indexCreator = new MongoPersistentEntityIndexCreator((MongoMappingContext) mappingContext,
-					(collectionName) -> IndexOperationsAdapter.blocking(indexOps(collectionName)));
-			eventPublisher = new MongoMappingEventPublisher(indexCreator);
-			if (mappingContext instanceof ApplicationEventPublisherAware) {
-				((ApplicationEventPublisherAware) mappingContext).setApplicationEventPublisher(eventPublisher);
-			}
+		// We create indexes based on mapping events
+		if (this.mappingContext instanceof MongoMappingContext) {
+
+			MongoMappingContext mongoMappingContext = (MongoMappingContext) this.mappingContext;
+			this.indexCreator = new ReactiveMongoPersistentEntityIndexCreator(mongoMappingContext, this::indexOps);
+			this.eventPublisher = new MongoMappingEventPublisher(this.indexCreatorListener);
+
+			mongoMappingContext.setApplicationEventPublisher(this.eventPublisher);
+			this.mappingContext.getPersistentEntities()
+					.forEach(entity -> onCheckForIndexes(entity, subscriptionExceptionHandler));
 		}
 	}
 
@@ -254,10 +275,22 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		this.updateMapper = that.updateMapper;
 		this.schemaMapper = that.schemaMapper;
 		this.projectionFactory = that.projectionFactory;
-
+		this.indexCreator = that.indexCreator;
+		this.indexCreatorListener = that.indexCreatorListener;
 		this.mappingContext = that.mappingContext;
 	}
 
+	private void onCheckForIndexes(MongoPersistentEntity<?> entity, Consumer<Throwable> subscriptionExceptionHandler) {
+
+		if (indexCreator != null) {
+			indexCreator.checkForIndexes(entity).subscribe(v -> {}, subscriptionExceptionHandler);
+		}
+	}
+
+	private static void handleSubscriptionException(Throwable t) {
+		LOGGER.error("Unexpected exception during asynchronous execution", t);
+	}
+
 	/**
 	 * Configures the {@link WriteResultChecking} to be used with the template. Setting {@literal null} will reset the
 	 * default of {@link ReactiveMongoTemplate#DEFAULT_WRITE_RESULT_CHECKING}.
@@ -289,7 +322,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	}
 
 	/**
-	 * Used by @{link {@link #prepareCollection(MongoCollection)} to set the {@link ReadPreference} before any operations
+	 * Used by {@link {@link #prepareCollection(MongoCollection)} to set the {@link ReadPreference} before any operations
 	 * are performed.
 	 *
 	 * @param readPreference
@@ -316,26 +349,28 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	}
 
 	/**
-	 * Inspects the given {@link ApplicationContext} for {@link MongoPersistentEntityIndexCreator} and those in turn if
-	 * they were registered for the current {@link MappingContext}. If no creator for the current {@link MappingContext}
-	 * can be found we manually add the internally created one as {@link ApplicationListener} to make sure indexes get
-	 * created appropriately for entity types persisted through this {@link ReactiveMongoTemplate} instance.
+	 * Inspects the given {@link ApplicationContext} for {@link ReactiveMongoPersistentEntityIndexCreator} and those in
+	 * turn if they were registered for the current {@link MappingContext}. If no creator for the current
+	 * {@link MappingContext} can be found we manually add the internally created one as {@link ApplicationListener} to
+	 * make sure indexes get created appropriately for entity types persisted through this {@link ReactiveMongoTemplate}
+	 * instance.
 	 *
 	 * @param context must not be {@literal null}.
 	 */
 	private void prepareIndexCreator(ApplicationContext context) {
 
-		String[] indexCreators = context.getBeanNamesForType(MongoPersistentEntityIndexCreator.class);
+		String[] indexCreators = context.getBeanNamesForType(ReactiveMongoPersistentEntityIndexCreator.class);
 
 		for (String creator : indexCreators) {
-			MongoPersistentEntityIndexCreator creatorBean = context.getBean(creator, MongoPersistentEntityIndexCreator.class);
+			ReactiveMongoPersistentEntityIndexCreator creatorBean = context.getBean(creator,
+					ReactiveMongoPersistentEntityIndexCreator.class);
 			if (creatorBean.isIndexCreatorFor(mappingContext)) {
 				return;
 			}
 		}
 
 		if (context instanceof ConfigurableApplicationContext) {
-			((ConfigurableApplicationContext) context).addApplicationListener(indexCreator);
+			((ConfigurableApplicationContext) context).addApplicationListener(indexCreatorListener);
 		}
 	}
 
@@ -3149,4 +3184,25 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 			return delegate.getMongoDatabase();
 		}
 	}
+
+	@RequiredArgsConstructor
+	class IndexCreatorEventListener implements ApplicationListener<MappingContextEvent<?, ?>> {
+
+		final Consumer<Throwable> subscriptionExceptionHandler;
+
+		@Override
+		public void onApplicationEvent(MappingContextEvent<?, ?> event) {
+
+			if (!event.wasEmittedBy(mappingContext)) {
+				return;
+			}
+
+			PersistentEntity<?, ?> entity = event.getPersistentEntity();
+
+			// Double check type as Spring infrastructure does not consider nested generics
+			if (entity instanceof MongoPersistentEntity) {
+				onCheckForIndexes((MongoPersistentEntity<?>) entity, subscriptionExceptionHandler);
+			}
+		}
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/index/IndexOperationsProvider.java

@@ -13,16 +13,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.springframework.data.mongodb.core.index;
 
 /**
- * TODO: Revisit for a better pattern.
- *
  * @author Mark Paluch
  * @author Jens Schauder
  * @since 2.0
  */
+@FunctionalInterface
 public interface IndexOperationsProvider {
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/index/MongoMappingEventPublisher.java

@@ -18,6 +18,7 @@ package org.springframework.data.mongodb.core.index;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationListener;
 import org.springframework.data.mapping.context.MappingContextEvent;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
@@ -38,7 +39,19 @@ import org.springframework.util.Assert;
  */
 public class MongoMappingEventPublisher implements ApplicationEventPublisher {
 
-	private final MongoPersistentEntityIndexCreator indexCreator;
+	private final ApplicationListener<MappingContextEvent<?, ?>> indexCreator;
+
+	/**
+	 * Creates a new {@link MongoMappingEventPublisher} for the given {@link ApplicationListener}.
+	 *
+	 * @param indexCreator must not be {@literal null}.
+	 * @since 2.1
+	 */
+	public MongoMappingEventPublisher(ApplicationListener<MappingContextEvent<?, ?>> indexCreator) {
+
+		Assert.notNull(indexCreator, "ApplicationListener must not be null!");
+		this.indexCreator = indexCreator;
+	}
 
 	/**
 	 * Creates a new {@link MongoMappingEventPublisher} for the given {@link MongoPersistentEntityIndexCreator}.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/index/ReactiveIndexOperationsProvider.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2018 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
+ *
+ *      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 org.springframework.data.mongodb.core.index;
+
+/**
+ * Provider interface to obtain {@link ReactiveIndexOperations} by MongoDB collection name.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+@FunctionalInterface
+public interface ReactiveIndexOperationsProvider {
+
+	/**
+	 * Returns the operations that can be performed on indexes.
+	 *
+	 * @param collectionName name of the MongoDB collection, must not be {@literal null}.
+	 * @return index operations on the named collection
+	 */
+	ReactiveIndexOperations indexOps(String collectionName);
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/index/ReactiveMongoPersistentEntityIndexCreator.java

@@ -0,0 +1,186 @@
+/*
+ * Copyright 2018 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
+ *
+ *      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 org.springframework.data.mongodb.core.index;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.mongodb.UncategorizedMongoDbException;
+import org.springframework.data.mongodb.core.index.MongoPersistentEntityIndexResolver.IndexDefinitionHolder;
+import org.springframework.data.mongodb.core.mapping.Document;
+import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
+import org.springframework.data.mongodb.util.MongoDbErrorCodes;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+import com.mongodb.MongoException;
+
+/**
+ * Component that inspects {@link MongoPersistentEntity} instances contained in the given {@link MongoMappingContext}
+ * for indexing metadata and ensures the indexes to be available using reactive infrastructure.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public class ReactiveMongoPersistentEntityIndexCreator {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveMongoPersistentEntityIndexCreator.class);
+
+	private final Map<Class<?>, Boolean> classesSeen = new ConcurrentHashMap<Class<?>, Boolean>();
+	private final MongoMappingContext mappingContext;
+	private final ReactiveIndexOperationsProvider operationsProvider;
+	private final IndexResolver indexResolver;
+
+	/**
+	 * Creates a new {@link ReactiveMongoPersistentEntityIndexCreator} for the given {@link MongoMappingContext},
+	 * {@link ReactiveIndexOperationsProvider}.
+	 *
+	 * @param mappingContext must not be {@literal null}.
+	 * @param operationsProvider must not be {@literal null}.
+	 */
+	public ReactiveMongoPersistentEntityIndexCreator(MongoMappingContext mappingContext,
+			ReactiveIndexOperationsProvider operationsProvider) {
+		this(mappingContext, operationsProvider, new MongoPersistentEntityIndexResolver(mappingContext));
+	}
+
+	/**
+	 * Creates a new {@link ReactiveMongoPersistentEntityIndexCreator} for the given {@link MongoMappingContext},
+	 * {@link ReactiveIndexOperationsProvider}, and {@link IndexResolver}.
+	 *
+	 * @param mappingContext must not be {@literal null}.
+	 * @param operationsProvider must not be {@literal null}.
+	 * @param indexResolver must not be {@literal null}.
+	 */
+	public ReactiveMongoPersistentEntityIndexCreator(MongoMappingContext mappingContext,
+			ReactiveIndexOperationsProvider operationsProvider, IndexResolver indexResolver) {
+
+		Assert.notNull(mappingContext, "MongoMappingContext must not be null!");
+		Assert.notNull(operationsProvider, "ReactiveIndexOperations must not be null!");
+		Assert.notNull(indexResolver, "IndexResolver must not be null!");
+
+		this.mappingContext = mappingContext;
+		this.operationsProvider = operationsProvider;
+		this.indexResolver = indexResolver;
+	}
+
+	/**
+	 * Returns whether the current index creator was registered for the given {@link MappingContext}.
+	 *
+	 * @param context
+	 * @return
+	 */
+	public boolean isIndexCreatorFor(MappingContext<?, ?> context) {
+		return this.mappingContext.equals(context);
+	}
+
+	/**
+	 * Inspect entities for index creation.
+	 *
+	 * @return a {@link Mono} that completes without value after indexes were created.
+	 */
+	public Mono<Void> checkForIndexes(MongoPersistentEntity<?> entity) {
+
+		Class<?> type = entity.getType();
+
+		if (!classesSeen.containsKey(type)) {
+
+			if (this.classesSeen.put(type, Boolean.TRUE) == null) {
+
+				if (LOGGER.isDebugEnabled()) {
+					LOGGER.debug("Analyzing class " + type + " for index information.");
+				}
+
+				return checkForAndCreateIndexes(entity);
+			}
+		}
+
+		return Mono.empty();
+	}
+
+	private Mono<Void> checkForAndCreateIndexes(MongoPersistentEntity<?> entity) {
+
+		List<Mono<?>> publishers = new ArrayList<>();
+
+		if (entity.isAnnotationPresent(Document.class)) {
+			for (IndexDefinitionHolder indexToCreate : indexResolver.resolveIndexFor(entity.getTypeInformation())) {
+				publishers.add(createIndex(indexToCreate));
+			}
+		}
+
+		return publishers.isEmpty() ? Mono.empty() : Flux.merge(publishers).then();
+	}
+
+	Mono<String> createIndex(IndexDefinitionHolder indexDefinition) {
+
+		return operationsProvider.indexOps(indexDefinition.getCollection()).ensureIndex(indexDefinition) //
+				.onErrorResume(ReactiveMongoPersistentEntityIndexCreator::isDataIntegrityViolation,
+						e -> translateException(e, indexDefinition));
+
+	}
+
+	private Mono<? extends String> translateException(Throwable e, IndexDefinitionHolder indexDefinition) {
+
+		Mono<IndexInfo> existingIndex = fetchIndexInformation(indexDefinition);
+
+		Mono<String> defaultError = Mono.error(new DataIntegrityViolationException(
+				String.format("Cannot create index for '%s' in collection '%s' with keys '%s' and options '%s'.",
+						indexDefinition.getPath(), indexDefinition.getCollection(), indexDefinition.getIndexKeys(),
+						indexDefinition.getIndexOptions()),
+				e.getCause()));
+
+		return existingIndex.flatMap(it -> {
+			return Mono.<String> error(new DataIntegrityViolationException(
+					String.format("Index already defined as '%s'.", indexDefinition.getPath()), e.getCause()));
+		}).switchIfEmpty(defaultError);
+	}
+
+	private Mono<IndexInfo> fetchIndexInformation(IndexDefinitionHolder indexDefinition) {
+
+		Object indexNameToLookUp = indexDefinition.getIndexOptions().get("name");
+
+		Flux<IndexInfo> existingIndexes = operationsProvider.indexOps(indexDefinition.getCollection()).getIndexInfo();
+
+		return existingIndexes //
+				.filter(indexInfo -> ObjectUtils.nullSafeEquals(indexNameToLookUp, indexInfo.getName())) //
+				.next() //
+				.doOnError(e -> {
+					LOGGER.debug(
+							String.format("Failed to load index information for collection '%s'.", indexDefinition.getCollection()),
+							e);
+				});
+	}
+
+	private static boolean isDataIntegrityViolation(Throwable t) {
+
+		if (t instanceof UncategorizedMongoDbException) {
+
+			return t.getCause() instanceof MongoException
+					&& MongoDbErrorCodes.isDataIntegrityViolationCode(((MongoException) t.getCause()).getCode());
+		}
+
+		return false;
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateIndexTests.java

@@ -23,7 +23,11 @@ import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
+import org.assertj.core.api.Assertions;
 import org.bson.Document;
 import org.junit.After;
 import org.junit.Before;
@@ -32,11 +36,13 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.mongodb.core.index.Index;
 import org.springframework.data.mongodb.core.index.IndexField;
 import org.springframework.data.mongodb.core.index.IndexInfo;
+import org.springframework.data.mongodb.core.index.Indexed;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -60,7 +66,9 @@ public class ReactiveMongoTemplateIndexTests {
 
 	@Before
 	public void setUp() {
+
 		StepVerifier.create(template.dropCollection(Person.class)).verifyComplete();
+		StepVerifier.create(template.dropCollection("indexfail")).verifyComplete();
 	}
 
 	@After
@@ -175,6 +183,39 @@ public class ReactiveMongoTemplateIndexTests {
 				}).verifyComplete();
 	}
 
+	@Test // DATAMONGO-1928
+	public void shouldCreateIndexOnAccess() {
+
+		template.findAll(IndexedSample.class) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		template.indexOps(IndexedSample.class).getIndexInfo() //
+				.as(StepVerifier::create) //
+				.expectNextCount(2) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-1928
+	public void indexCreationShouldFail() throws InterruptedException {
+
+		String command = "db.indexfail" + ".createIndex({'field':1}, {'name':'foo', 'unique':true, 'sparse':true}), 1";
+
+		StepVerifier.create(factory.getMongoDatabase().runCommand(new org.bson.Document("eval", command))) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		BlockingQueue<Throwable> queue = new LinkedBlockingQueue<>();
+
+		ReactiveMongoTemplate template = new ReactiveMongoTemplate(factory, this.template.getConverter(), queue::add);
+
+		template.findAll(IndexCreationShouldFail.class).subscribe();
+
+		Throwable failure = queue.poll(10, TimeUnit.SECONDS);
+
+		Assertions.assertThat(failure).isNotNull().isInstanceOf(DataIntegrityViolationException.class);
+	}
+
 	@Data
 	static class Sample {
 
@@ -188,4 +229,20 @@ public class ReactiveMongoTemplateIndexTests {
 			this.field = field;
 		}
 	}
+
+	@Data
+	@org.springframework.data.mongodb.core.mapping.Document
+	static class IndexedSample {
+
+		@Id String id;
+		@Indexed String field;
+	}
+
+	@Data
+	@org.springframework.data.mongodb.core.mapping.Document("indexfail")
+	static class IndexCreationShouldFail {
+
+		@Id String id;
+		@Indexed(name = "foo") String field;
+	}
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/index/ReactiveMongoPersistentEntityIndexCreatorUnitTests.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2018 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
+ *
+ *      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 org.springframework.data.mongodb.core.index;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
+import org.springframework.data.mongodb.core.MongoExceptionTranslator;
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.data.mongodb.core.mapping.Document;
+import org.springframework.data.mongodb.core.mapping.Field;
+import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+
+import com.mongodb.MongoException;
+import com.mongodb.client.model.IndexOptions;
+import com.mongodb.reactivestreams.client.MongoCollection;
+import com.mongodb.reactivestreams.client.MongoDatabase;
+
+/**
+ * Unit tests for {@link ReactiveMongoPersistentEntityIndexCreator}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ReactiveMongoPersistentEntityIndexCreatorUnitTests {
+
+	ReactiveIndexOperations indexOperations;
+
+	@Mock ReactiveMongoDatabaseFactory factory;
+	@Mock MongoDatabase db;
+	@Mock MongoCollection<org.bson.Document> collection;
+
+	ArgumentCaptor<org.bson.Document> keysCaptor;
+	ArgumentCaptor<IndexOptions> optionsCaptor;
+	ArgumentCaptor<String> collectionCaptor;
+
+	@Before
+	@SuppressWarnings("unchecked")
+	public void setUp() {
+
+		when(factory.getExceptionTranslator()).thenReturn(new MongoExceptionTranslator());
+		when(factory.getMongoDatabase()).thenReturn(db);
+		when(db.getCollection(any(), any(Class.class))).thenReturn(collection);
+
+		indexOperations = new ReactiveMongoTemplate(factory).indexOps("foo");
+
+		keysCaptor = ArgumentCaptor.forClass(org.bson.Document.class);
+		optionsCaptor = ArgumentCaptor.forClass(IndexOptions.class);
+		collectionCaptor = ArgumentCaptor.forClass(String.class);
+
+		when(collection.createIndex(keysCaptor.capture(), optionsCaptor.capture())).thenReturn(Mono.just("OK"));
+	}
+
+	@Test // DATAMONGO-1928
+	public void buildsIndexDefinitionUsingFieldName() {
+
+		MongoMappingContext mappingContext = prepareMappingContext(Person.class);
+
+		Mono<Void> publisher = checkForIndexes(mappingContext);
+
+		verifyZeroInteractions(collection);
+
+		publisher.as(StepVerifier::create).verifyComplete();
+
+		assertThat(keysCaptor.getValue()).isNotNull().containsKey("fieldname");
+		assertThat(optionsCaptor.getValue().getName()).isEqualTo("indexName");
+		assertThat(optionsCaptor.getValue().isBackground()).isFalse();
+		assertThat(optionsCaptor.getValue().getExpireAfter(TimeUnit.SECONDS)).isNull();
+	}
+
+	@Test // DATAMONGO-1928
+	public void createIndexShouldUsePersistenceExceptionTranslatorForNonDataIntegrityConcerns() {
+
+		when(collection.createIndex(any(org.bson.Document.class), any(IndexOptions.class)))
+				.thenReturn(Mono.error(new MongoException(6, "HostUnreachable")));
+
+		MongoMappingContext mappingContext = prepareMappingContext(Person.class);
+
+		Mono<Void> publisher = checkForIndexes(mappingContext);
+
+		publisher.as(StepVerifier::create).expectError(DataAccessResourceFailureException.class).verify();
+	}
+
+	@Test // DATAMONGO-1928
+	public void createIndexShouldNotConvertUnknownExceptionTypes() {
+
+		when(collection.createIndex(any(org.bson.Document.class), any(IndexOptions.class)))
+				.thenReturn(Mono.error(new ClassCastException("o_O")));
+
+		MongoMappingContext mappingContext = prepareMappingContext(Person.class);
+
+		Mono<Void> publisher = checkForIndexes(mappingContext);
+
+		publisher.as(StepVerifier::create).expectError(ClassCastException.class).verify();
+	}
+
+	private static MongoMappingContext prepareMappingContext(Class<?> type) {
+
+		MongoMappingContext mappingContext = new MongoMappingContext();
+		mappingContext.setInitialEntitySet(Collections.singleton(type));
+		mappingContext.initialize();
+
+		return mappingContext;
+	}
+
+	private Mono<Void> checkForIndexes(MongoMappingContext mappingContext) {
+
+		return new ReactiveMongoPersistentEntityIndexCreator(mappingContext, it -> indexOperations)
+				.checkForIndexes(mappingContext.getRequiredPersistentEntity(Person.class));
+	}
+
+	@Document
+	static class Person {
+
+		@Indexed(name = "indexName") //
+		@Field("fieldname") //
+		String field;
+
+	}
+}