Support errors.tolerance

Scenarios covered:

Source

  - Missing / invalid / not found Resume Tokens (Integration test with a mocked offsetStorageReader)
  - Poison pill message - invalid schema

Sink

  - Poison pill messages - Invalid Key / Values types & invalid documents
  - Errors thrown by PostProcessors
  - Debezium CDC handler errors / poison pills

KAFKA-105

Author: rozza

CHANGELOG.md

@@ -21,6 +21,7 @@
   - [KAFKA-78](https://jira.mongodb.org/browse/KAFKA-78) Added dead letter queue support for the source connector.
   - [KAFKA-157](https://jira.mongodb.org/browse/KAFKA-157) Improved error message for business key errors.
   - [KAFKA-155](https://jira.mongodb.org/browse/KAFKA-155) Fix business key update strategies to use dot notation for filters
+  - [KAFKA-105](https://jira.mongodb.org/browse/KAFKA-105) Improve `errors.tolerance=all` support in the sink and source connectors.
 
 
 ## 1.2.0

src/integrationTest/java/com/mongodb/kafka/connect/log/LogCapture.java

@@ -45,6 +45,10 @@ public List<LoggingEvent> getEvents() {
     return new ArrayList<>(loggingEvents);
   }
 
+  public void reset() {
+    loggingEvents.clear();
+  }
+
   @Override
   public void close() {
     logger.removeAppender(APPENDER_NAME);

src/integrationTest/java/com/mongodb/kafka/connect/mongodb/MongoKafkaTestCase.java

@@ -157,28 +157,30 @@ public MongoCollection<Document> getAndCreateCollection() {
 
   private static final String SIMPLE_DOCUMENT = "{_id: %s}";
 
+  public List<Document> createDocuments(final IntStream stream) {
+    return createDocuments(stream, SIMPLE_DOCUMENT);
+  }
+
+  public List<Document> createDocuments(final IntStream stream, final String json) {
+    return stream.mapToObj(i -> Document.parse(format(json, i))).collect(toList());
+  }
+
   public List<Document> insertMany(
       final IntStream stream, final MongoCollection<?>... collections) {
     return insertMany(stream, SIMPLE_DOCUMENT, collections);
   }
 
   public List<Document> insertMany(
       final IntStream stream, final String json, final MongoCollection<?>... collections) {
-    List<Document> docs = stream.mapToObj(i -> Document.parse(format(json, i))).collect(toList());
+    List<Document> docs = createDocuments(stream, json);
     for (MongoCollection<?> c : collections) {
-      LOGGER.debug("Inserting into {} ", c.getNamespace().getFullName());
+      LOGGER.debug("Inserting {} documents into {} ", docs.size(), c.getNamespace().getFullName());
       c.withDocumentClass(Document.class).insertMany(docs);
     }
     return docs;
   }
 
-  public void assertCollection(
-      final MongoCollection<BsonDocument> source, final MongoCollection<BsonDocument> destination) {
-    assertCollection(source.find().into(new ArrayList<>()), destination);
-  }
-
-  public void assertCollection(
-      final List<BsonDocument> expected, final MongoCollection<BsonDocument> destination) {
+  public <T> void assertCollection(final List<T> expected, final MongoCollection<T> destination) {
     int counter = 0;
     int retryCount = 0;
     while (retryCount < DEFAULT_MAX_RETRIES) {