CC-41565/41566/41568/41569: stop leaking Kafka record content into logs (v1.16.x backport)

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

@@ -107,7 +107,13 @@ private <T> Optional<T> tryProcess(final Supplier<Optional<T>> supplier) {
       exception = e;
       if (config.logErrors()) {
         LOGGER.error(
-            "Unable to process record in topic:{} at partition:{}, offset:{}",
+            "Unable to process record in topic:{} at partition:{}, offset:{}. Cause: {}",
+            sinkRecord.topic(),
+            sinkRecord.kafkaPartition(),
+            sinkRecord.kafkaOffset(),
+            e.getClass().getName());
+        LOGGER.debug(
+            "Unable to process record in topic:{} at partition:{}, offset:{} (full detail)",
             sinkRecord.topic(),
             sinkRecord.kafkaPartition(),
             sinkRecord.kafkaOffset(),

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

@@ -214,16 +214,29 @@ private void handleTolerableWriteException(
       final boolean logErrors,
       final boolean tolerateErrors) {
     if (e instanceof MongoBulkWriteException) {
+      MongoBulkWriteException bulkWriteException = (MongoBulkWriteException) e;
       AnalyzedBatchFailedWithBulkWriteException analyzedBatch =
           new AnalyzedBatchFailedWithBulkWriteException(
               batch,
               ordered,
-              (MongoBulkWriteException) e,
+              bulkWriteException,
               errorReporter,
               StartedMongoSinkTask::log);
       if (logErrors) {
         LOGGER.error(
+            "Failed to write some records into the sink: {} write error(s) with code(s) {}, "
+                + "{} write concern error(s). Exception: {}",
+            bulkWriteException.getWriteErrors().size(),
+            bulkWriteException.getWriteErrors().stream()
+                .map(writeError -> Integer.toString(writeError.getCode()))
+                .distinct()
+                .collect(Collectors.toList()),
+            bulkWriteException.getWriteConcernError() != null ? 1 : 0,
+            e.getClass().getName());
+        LOGGER.debug(
             "Failed to put into the sink some records, see log entries below for the details", e);
+        // Per-record breakdown: routes through the sanitized log() method below, so each entry is
+        // logged at ERROR without record content (full per-record detail stays at DEBUG).
         analyzedBatch.log();
       }
       if (tolerateErrors) {
@@ -244,6 +257,10 @@ private void handleTolerableWriteException(
   }
 
   private static void log(final Collection<SinkRecord> records, final RuntimeException e) {
-    LOGGER.error("Failed to put {} records into the sink", records.size(), e);
+    LOGGER.error(
+        "Failed to put {} records into the sink. Exception: {}",
+        records.size(),
+        e.getClass().getName());
+    LOGGER.debug("Failed to put {} records into the sink (full detail)", records.size(), e);
   }
 }

src/main/java/com/mongodb/kafka/connect/sink/cdc/mongodb/operations/OperationHelper.java

@@ -52,8 +52,8 @@ static BsonDocument getDocumentKey(final BsonDocument changeStreamDocument) {
     } else if (!changeStreamDocument.get(DOCUMENT_KEY).isDocument()) {
       throw new DataException(
           format(
-              "Unexpected %s field type, expecting a document but found `%s`",
-              DOCUMENT_KEY, changeStreamDocument.get(DOCUMENT_KEY)));
+              "Unexpected %s field type, expecting a document but found a value of type `%s`",
+              DOCUMENT_KEY, changeStreamDocument.get(DOCUMENT_KEY).getBsonType()));
     }
 
     return changeStreamDocument.getDocument(DOCUMENT_KEY);
@@ -69,8 +69,8 @@ static BsonDocument getFullDocument(final BsonDocument changeStreamDocument) {
     } else if (!changeStreamDocument.get(FULL_DOCUMENT).isDocument()) {
       throw new DataException(
           format(
-              "Unexpected %s field type, expecting a document but found `%s`",
-              FULL_DOCUMENT, changeStreamDocument.get(FULL_DOCUMENT)));
+              "Unexpected %s field type, expecting a document but found a value of type `%s`",
+              FULL_DOCUMENT, changeStreamDocument.get(FULL_DOCUMENT).getBsonType()));
     }
 
     return changeStreamDocument.getDocument(FULL_DOCUMENT);
@@ -82,8 +82,8 @@ static BsonDocument getUpdateDocument(final BsonDocument changeStreamDocument) {
     } else if (!changeStreamDocument.get(UPDATE_DESCRIPTION).isDocument()) {
       throw new DataException(
           format(
-              "Unexpected %s field type, expected a document found `%s`",
-              UPDATE_DESCRIPTION, changeStreamDocument.get(UPDATE_DESCRIPTION)));
+              "Unexpected %s field type, expected a document but found a value of type `%s`",
+              UPDATE_DESCRIPTION, changeStreamDocument.get(UPDATE_DESCRIPTION).getBsonType()));
     }
 
     BsonDocument updateDescription = changeStreamDocument.getDocument(UPDATE_DESCRIPTION);
@@ -101,35 +101,37 @@ static BsonDocument getUpdateDocument(final BsonDocument changeStreamDocument) {
     } else if (!updateDescription.get(UPDATED_FIELDS).isDocument()) {
       throw new DataException(
           format(
-              "Unexpected %s field type, expected a document but found `%s`",
-              UPDATE_DESCRIPTION, updateDescription));
+              "Unexpected %s.%s field type, expected a document but found a value of type `%s`",
+              UPDATE_DESCRIPTION,
+              UPDATED_FIELDS,
+              updateDescription.get(UPDATED_FIELDS).getBsonType()));
     }
 
     if (!updateDescription.containsKey(REMOVED_FIELDS)) {
       throw new DataException(format("Missing %s.%s field", UPDATE_DESCRIPTION, REMOVED_FIELDS));
     } else if (!updateDescription.get(REMOVED_FIELDS).isArray()) {
       throw new DataException(
           format(
-              "Unexpected %s field type, expected an array but found `%s`",
-              REMOVED_FIELDS, updateDescription.get(REMOVED_FIELDS)));
+              "Unexpected %s field type, expected an array but found a value of type `%s`",
+              REMOVED_FIELDS, updateDescription.get(REMOVED_FIELDS).getBsonType()));
     }
 
     if (updateDescription.containsKey(TRUNCATED_ARRAYS)
         && !updateDescription.get(TRUNCATED_ARRAYS).isArray()) {
       throw new DataException(
           format(
-              "Unexpected %s field type, expected an array but found `%s`",
+              "Unexpected %s field type, expected an array but found a value of type `%s`",
               TRUNCATED_ARRAYS,
-              updateDescription.get(TRUNCATED_ARRAYS)));
+              updateDescription.get(TRUNCATED_ARRAYS).getBsonType()));
     }
 
     if (updateDescription.containsKey(DISAMBIGUATED_PATHS)
         && !updateDescription.get(DISAMBIGUATED_PATHS).isDocument()) {
       throw new DataException(
           format(
-              "Unexpected %s field type, expected an array but found `%s`",
+              "Unexpected %s field type, expected an array but found a value of type `%s`",
               DISAMBIGUATED_PATHS,
-              updateDescription.get(DISAMBIGUATED_PATHS)));
+              updateDescription.get(DISAMBIGUATED_PATHS).getBsonType()));
     }
 
     BsonDocument updatedFields = updateDescription.getDocument(UPDATED_FIELDS);
@@ -139,8 +141,8 @@ static BsonDocument getUpdateDocument(final BsonDocument changeStreamDocument) {
       if (!removedField.isString()) {
         throw new DataException(
             format(
-                "Unexpected value type in %s, expected an string but found `%s`",
-                REMOVED_FIELDS, removedField));
+                "Unexpected value type in %s, expected a string but found a value of type `%s`",
+                REMOVED_FIELDS, removedField.getBsonType()));
       }
       unsetDocument.append(removedField.asString().getValue(), EMPTY_STRING);
     }

src/main/java/com/mongodb/kafka/connect/sink/processor/TimeseriesTimeFieldAutoConversion.java

@@ -78,9 +78,14 @@ Optional<BsonDateTime> tryToConvert(final Supplier<BsonDateTime> supplier) {
       return Optional.of(supplier.get());
     } catch (Exception e) {
       LOGGER.info(
+          "Failed to convert field `{}` to a valid date time, so leaving as is. Cause: {}",
+          fieldName,
+          e.getClass().getSimpleName());
+      LOGGER.trace(
           format(
               "Failed to convert field `%s` to a valid date time, so leaving as is: `%s`",
-              fieldName, e.getMessage()));
+              fieldName, e.getMessage()),
+          e);
       return Optional.empty();
     }
   }

src/main/java/com/mongodb/kafka/connect/sink/processor/id/strategy/UuidProvidedStrategy.java

@@ -50,7 +50,8 @@ public BsonValue generateId(final SinkDocument doc, final SinkRecord orig) {
           constructUuidObjectFromString(id.asString().getValue()), UuidRepresentation.STANDARD);
     }
 
-    throw new DataException(format("UUID cannot be constructed from provided value: `%s`", id));
+    throw new DataException(
+        format("UUID cannot be constructed from a provided value of type `%s`", id.getBsonType()));
   }
 
   private UUID constructUuidObjectFromString(final String uuid) {
@@ -67,6 +68,7 @@ private UUID constructUuidObjectFromString(final String uuid) {
       // ignore
     }
 
-    throw new DataException(format("UUID cannot be constructed from provided value: `%s`", uuid));
+    throw new DataException(
+        format("UUID cannot be constructed from the provided string value (length=%d)", uuid.length()));
   }
 }

src/main/java/com/mongodb/kafka/connect/util/Validators.java

@@ -168,10 +168,12 @@ public static ValidatorWithOperators errorCheckingPasswordValueValidator(
           try {
             consumer.accept((String) value);
           } catch (IllegalArgumentException e){
-            LOGGER.error(e.getMessage());
+            LOGGER.error("Invalid {} value: connection string could not be parsed", name);
+            LOGGER.debug("Connection string validation failure detail for {}", name, e);
             throw new ConfigException(name, redactedUrl, connectionUriErrorMessage);
           } catch (Exception e) {
-            throw new ConfigException(name, redactedUrl, e.getMessage());
+            LOGGER.debug("Unexpected error validating {}", name, e);
+            throw new ConfigException(name, redactedUrl, connectionUriErrorMessage);
           }
         }));
   }