diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java
index aa2284f7a..46ae61b50 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java
@@ -38,6 +38,7 @@ import com.datastax.driver.core.querybuilder.Select;
  * Reactive repository base implementation for Cassandra.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.0
  */
 public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassandraRepository<T, ID> {
@@ -94,27 +95,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
 
 		Assert.notNull(entityStream, "The given Publisher of entities must not be null");
 
-		return Flux.from(entityStream).flatMap(entity -> {
-			return operations.getReactiveCqlOperations().execute(createFullInsert(entity)).map(it -> entity);
-		});
-	}
-
-	private <S extends T> Insert createFullInsert(S entity) {
-
-		CassandraConverter converter = operations.getConverter();
-		CassandraPersistentEntity<?> persistentEntity = converter.getMappingContext()
-				.getRequiredPersistentEntity(entity.getClass());
-		Map<String, Object> toInsert = new LinkedHashMap<>();
-
-		converter.write(entity, toInsert, persistentEntity);
-
-		Insert insert = QueryBuilder.insertInto(persistentEntity.getTableName().toCql());
-
-		for (Entry<String, Object> entry : toInsert.entrySet()) {
-			insert.value(entry.getKey(), entry.getValue());
-		}
-
-		return insert;
+		return Flux.from(entityStream).flatMap(entity -> operations.getReactiveCqlOperations().execute(createFullInsert(entity)).map(it -> entity));
 	}
 
 	/* (non-Javadoc)
@@ -245,6 +226,17 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
 		return operations.deleteById(id, entityInformation.getJavaType()).then();
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteById(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Mono<Void> deleteById(Publisher<ID> publisher) {
+
+		Assert.notNull(publisher, "The given id must not be null");
+
+		return Mono.from(publisher).flatMap(id -> operations.deleteById(id, entityInformation.getJavaType())).then();
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.repository.reactive.ReactiveCrudRepository#delete(java.lang.Object)
 	 */
@@ -285,4 +277,22 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
 	public Mono<Void> deleteAll() {
 		return operations.truncate(entityInformation.getJavaType());
 	}
+
+	private <S extends T> Insert createFullInsert(S entity) {
+
+		CassandraConverter converter = operations.getConverter();
+		CassandraPersistentEntity<?> persistentEntity = converter.getMappingContext()
+				.getRequiredPersistentEntity(entity.getClass());
+		Map<String, Object> toInsert = new LinkedHashMap<>();
+
+		converter.write(entity, toInsert, persistentEntity);
+
+		Insert insert = QueryBuilder.insertInto(persistentEntity.getTableName().toCql());
+
+		for (Entry<String, Object> entry : toInsert.entrySet()) {
+			insert.value(entry.getKey(), entry.getValue());
+		}
+
+		return insert;
+	}
 }
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java
index d2d76f365..52a138542 100644
--- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java
@@ -44,6 +44,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  * Integration tests for {@link SimpleReactiveCassandraRepository}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration
@@ -127,8 +128,8 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
 		StepVerifier.create(repository.existsById(Mono.just(dave.getId()))).expectNext(true).verifyComplete();
 	}
 
-	@Test // DATACASS-335
-	public void existsByFluxOfIdShouldReturnTrueForExistingObject() {
+	@Test // DATACASS-462
+	public void existsByIdUsingFluxShouldReturnTrueForExistingObject() {
 
 		insertTestData();
 
@@ -163,7 +164,7 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
 	}
 
 	@Test // DATACASS-462
-	public void findByIdByFluxOfIdShouldReturnTrueForExistingObject() {
+	public void findByIdUsingFluxShouldReturnTrueForExistingObject() {
 
 		insertTestData();
 
@@ -321,14 +322,39 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
 	@Test // DATACASS-335
 	public void deleteByIdShouldRemoveEntity() {
 
+		insertTestData();
+
 		StepVerifier.create(repository.deleteById(dave.getId())).verifyComplete();
 
 		StepVerifier.create(repository.findById(dave.getId())).expectNextCount(0).verifyComplete();
 	}
 
+	@Test // DATACASS-462
+	public void deleteByIdUsingMonoShouldRemoveEntity() {
+
+		insertTestData();
+
+		StepVerifier.create(repository.deleteById(Mono.just(dave.getId()))).verifyComplete();
+
+		StepVerifier.create(repository.existsById(dave.getId())).expectNext(false).verifyComplete();
+	}
+
+	@Test // DATACASS-462
+	public void deleteByIdUsingFluxShouldRemoveFirstEntity() {
+
+		insertTestData();
+
+		StepVerifier.create(repository.deleteById(Flux.just(dave.getId(), oliver.getId()))).verifyComplete();
+
+		StepVerifier.create(repository.existsById(dave.getId())).expectNext(false).verifyComplete();
+		StepVerifier.create(repository.existsById(oliver.getId())).expectNext(true).verifyComplete();
+	}
+
 	@Test // DATACASS-335
 	public void deleteShouldRemoveEntity() {
 
+		insertTestData();
+
 		StepVerifier.create(repository.delete(dave)).verifyComplete();
 
 		StepVerifier.create(repository.findById(dave.getId())).expectNextCount(0).verifyComplete();
@@ -337,6 +363,8 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
 	@Test // DATACASS-335
 	public void deleteIterableOfEntitiesShouldRemoveEntities() {
 
+		insertTestData();
+
 		StepVerifier.create(repository.deleteAll(Arrays.asList(dave, boyd))).verifyComplete();
 
 		StepVerifier.create(repository.findById(boyd.getId())).expectNextCount(0).verifyComplete();
@@ -345,6 +373,8 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
 	@Test // DATACASS-335
 	public void deletePublisherOfEntitiesShouldRemoveEntities() {
 
+		insertTestData();
+
 		StepVerifier.create(repository.deleteAll(Flux.just(dave, boyd))).verifyComplete();
 
 		StepVerifier.create(repository.findById(boyd.getId())).expectNextCount(0).verifyComplete();