Changed max.num.retries default to 1.

A safer default, especially as the driver now has retryable writes.

KAFKA-106

Author: rozza

CHANGELOG.md

@@ -22,7 +22,7 @@
   - [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.
-
+  - [KAFKA-106](https://jira.mongodb.org/browse/KAFKA-106) Changed `max.num.retries` default to 1. A safer default especially as the driver now has retryable writes.
 
 ## 1.2.0
   - [KAFKA-92](https://jira.mongodb.org/browse/KAFKA-92) Allow the Sink connector to use multiple tasks.

config/MongoSinkConnector.properties

@@ -13,7 +13,7 @@ value.converter.schema.registry.url=http://localhost:8081
 connection.uri=mongodb://mongo1:27017,mongo2:27017,mongo3:27017
 database=test
 collection=sink
-max.num.retries=3
+max.num.retries=1
 retries.defer.timeout=5000
 
 

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkTask.java

@@ -146,6 +146,7 @@ public void put(final Collection<SinkRecord> records) {
                       }
                     }
                   });
+          resetRemainingRetriesForTopic(topicConfig);
         });
   }
 
@@ -206,11 +207,10 @@ private void processSinkRecords(final MongoSinkTopicConfig config, final List<Si
       }
     } catch (MongoException e) {
       LOGGER.warn(
-          "Writing {} document(s) into collection [{}] failed -> remaining retries ({})",
+          "Writing {} document(s) into collection [{}] failed.",
           writeModels.size(),
-          config.getNamespace().getFullName(),
-          getRemainingRetriesForTopic(config.getTopic()).get());
-      checkRetriableException(config, e);
+          config.getNamespace().getFullName());
+      handleMongoException(config, e);
     } catch (Exception e) {
       if (!config.tolerateErrors()) {
         throw new DataException("Failed to write mongodb documents", e);
@@ -228,7 +228,12 @@ private AtomicInteger getRemainingRetriesForTopic(final String topic) {
     return remainingRetriesTopicMap.get(topic);
   }
 
-  private void checkRetriableException(final MongoSinkTopicConfig config, final MongoException e) {
+  private void resetRemainingRetriesForTopic(final MongoSinkTopicConfig topicConfig) {
+    getRemainingRetriesForTopic(topicConfig.getTopic())
+        .set(topicConfig.getInt(MAX_NUM_RETRIES_CONFIG));
+  }
+
+  private void handleMongoException(final MongoSinkTopicConfig config, final MongoException e) {
     if (getRemainingRetriesForTopic(config.getTopic()).decrementAndGet() <= 0) {
       if (config.logErrors()) {
         LOGGER.error("Error on mongodb operation", e);

src/main/java/com/mongodb/kafka/connect/sink/MongoSinkTopicConfig.java

@@ -103,7 +103,7 @@ public String value() {
   private static final String MAX_NUM_RETRIES_DISPLAY = "Max number of retries";
   private static final String MAX_NUM_RETRIES_DOC =
       "How often a retry should be done on write errors";
-  private static final int MAX_NUM_RETRIES_DEFAULT = 3;
+  private static final int MAX_NUM_RETRIES_DEFAULT = 1;
 
   public static final String RETRIES_DEFER_TIMEOUT_CONFIG = "retries.defer.timeout";
   private static final String RETRIES_DEFER_TIMEOUT_DISPLAY = "Retry defer timeout";