From 8cbb8eb63b0e3f088a595f7b8a04e864c9129f8d Mon Sep 17 00:00:00 2001
From: Lev Zemlyanov <lev@confluent.io>
Date: Thu, 5 Sep 2019 14:30:24 -0700
Subject: [PATCH] DBZ-1464 Fixed NPE when op is not u/i/d

Signed-off-by: Lev Zemlyanov <lev@confluent.io>
---
 .../java/io/debezium/connector/mongodb/RecordMakers.java     | 4 ++++
 .../main/java/io/debezium/connector/mongodb/Replicator.java  | 5 +++++
 2 files changed, 9 insertions(+)

diff --git a/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/RecordMakers.java b/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/RecordMakers.java
index e303880f3..f98171401 100644
--- a/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/RecordMakers.java
+++ b/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/RecordMakers.java
@@ -94,6 +94,10 @@ public RecordsForCollection forCollection(CollectionId collectionId) {
         });
     }
 
+    public static boolean isValidOperation(String operation) {
+        return operationLiterals.containsKey(operation);
+    }
+
     /**
      * A record producer for a given collection.
      */
diff --git a/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/Replicator.java b/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/Replicator.java
index ce5c0f23c..9e5909e85 100644
--- a/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/Replicator.java
+++ b/debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/Replicator.java
@@ -544,6 +544,11 @@ protected boolean handleOplogEvent(ServerAddress primaryAddress, Document event)
             }
             return true;
         }
+        if (!RecordMakers.isValidOperation(event.getString("op"))) {
+            // the op is not insert/update/delete
+            logger.debug("Skipping event with \"op={}\"", event.getString("op"));
+            return true;
+        }
         int delimIndex = ns.indexOf('.');
         if (delimIndex > 0) {
             assert (delimIndex + 1) < ns.length();