Only drop collection for "drop" command #79

richardwilly98 · May 18, 2013 · bbe1d0d · bbe1d0d
1 parent 1070857
commit bbe1d0d
Showing 1 changed file with 16 additions and 11 deletions.
diff --git a/src/main/java/org/elasticsearch/river/mongodb/MongoDBRiver.java b/src/main/java/org/elasticsearch/river/mongodb/MongoDBRiver.java
@@ -143,6 +143,7 @@ public class MongoDBRiver extends AbstractRiverComponent implements River {
 	public final static String OPLOG_INSERT_OPERATION = "i";
 	public final static String OPLOG_DELETE_OPERATION = "d";
 	public final static String OPLOG_COMMAND_OPERATION = "c";
+	public final static String OPLOG_DROP_COMMAND_OPERATION = "drop";
 	public final static String OPLOG_TIMESTAMP = "ts";
 	public final static String GRIDFS_FILES_SUFFIX = ".files";
 	public final static String GRIDFS_CHUNKS_SUFFIX = ".chunks";
@@ -775,17 +776,21 @@ private BSONTimestamp updateBulkRequest(final BulkRequestBuilder bulk,
 				}
 				if (OPLOG_COMMAND_OPERATION.equals(operation)) {
 					if (dropCollection) {
-						logger.info("Drop collection request [{}], [{}]", index,
-								type);
-						bulk.request().requests().clear();
-						client.admin().indices().prepareDeleteMapping(index)
-								.setType(type).execute().actionGet();
-						deletedDocuments = 0;
-						updatedDocuments = 0;
-						insertedDocuments = 0;
-						logger.info(
-								"Delete request for index / type [{}] [{}] successfully executed.",
-								index, type);
+						if (data.containsKey(OPLOG_DROP_COMMAND_OPERATION) && data.get(OPLOG_DROP_COMMAND_OPERATION).equals(mongoCollection)) {
+							logger.info("Drop collection request [{}], [{}]", index,
+									type);
+							bulk.request().requests().clear();
+							client.admin().indices().prepareDeleteMapping(index)
+									.setType(type).execute().actionGet();
+							deletedDocuments = 0;
+							updatedDocuments = 0;
+							insertedDocuments = 0;
+							logger.info(
+									"Delete request for index / type [{}] [{}] successfully executed.",
+									index, type);
+						} else {
+							logger.debug("Database command {}", data);
+						}
 					} else {
 						logger.info("Ignore drop collection request [{}], [{}]. The option has been disabled.", index,
 								type);