Defensively handle the heartbeat manager

Also ensure its reset on restart.

KAFKA-161

Author: rozza

src/main/java/com/mongodb/kafka/connect/source/MongoSourceTask.java

@@ -165,6 +165,7 @@ public void start(final Map<String, String> props) {
       throw new ConnectException("Failed to start new task", e);
     }
 
+    heartbeatManager = null;
     createPartitionMap(sourceConfig, true);
 
     mongoClient =
@@ -210,7 +211,10 @@ public List<SourceRecord> poll() {
         if (!sourceRecords.isEmpty()) {
           return sourceRecords;
         }
-        return heartbeatManager.heartbeat().map(Collections::singletonList).orElse(null);
+        if (heartbeatManager != null) {
+          return heartbeatManager.heartbeat().map(Collections::singletonList).orElse(null);
+        }
+        return null;
       } else {
         BsonDocument changeStreamDocument = next.get();