Normalization logs: remove json parse warnings (#34978)

Author: edgao

airbyte-cdk/java/airbyte-cdk/README.md

@@ -166,6 +166,7 @@ MavenLocal debugging steps:
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.21.2  | 2024-02-20 | [\#34978](https://github.com/airbytehq/airbyte/pull/34978) | Reduce log noise in NormalizationLogParser.                                                                                                                    |
 | 0.21.1  | 2024-02-20 | [\#35199](https://github.com/airbytehq/airbyte/pull/35199) | Add thread names to the logs.                                                                                                                                  |
 | 0.21.0  | 2024-02-16 | [\#35314](https://github.com/airbytehq/airbyte/pull/35314) | Delete S3StreamCopier classes. These have been superseded by the async destinations framework.                                                                 |
 | 0.20.9  | 2024-02-15 | [\#35240](https://github.com/airbytehq/airbyte/pull/35240) | Make state emission to platform inside state manager itself.                                                                                                   |

airbyte-cdk/java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/destination/normalization/NormalizationLogParser.java

@@ -53,7 +53,7 @@ Stream<AirbyteMessage> toMessages(final String line) {
     if (Strings.isEmpty(line)) {
       return Stream.of(logMessage(Level.INFO, ""));
     }
-    final Optional<JsonNode> json = Jsons.tryDeserialize(line);
+    final Optional<JsonNode> json = Jsons.tryDeserializeWithoutWarn(line);
     if (json.isPresent()) {
       return jsonToMessage(json.get());
     } else {
@@ -96,7 +96,7 @@ private Stream<AirbyteMessage> jsonToMessage(final JsonNode jsonLine) {
        */
       final String logLevel = (jsonLine.hasNonNull("level")) ? jsonLine.get("level").asText() : "";
       String logMsg = jsonLine.hasNonNull("msg") ? jsonLine.get("msg").asText() : "";
-      Level level;
+      final Level level;
       switch (logLevel) {
         case "debug" -> level = Level.DEBUG;
         case "info" -> level = Level.INFO;
@@ -117,15 +117,15 @@ private Stream<AirbyteMessage> jsonToMessage(final JsonNode jsonLine) {
     }
   }
 
-  private static AirbyteMessage logMessage(Level level, String message) {
+  private static AirbyteMessage logMessage(final Level level, final String message) {
     return new AirbyteMessage()
         .withType(Type.LOG)
         .withLog(new AirbyteLogMessage()
             .withLevel(level)
             .withMessage(message));
   }
 
-  public static void main(String[] args) {
+  public static void main(final String[] args) {
     final NormalizationLogParser normalizationLogParser = new NormalizationLogParser();
     final Stream<AirbyteMessage> airbyteMessageStream =
         normalizationLogParser.create(new BufferedReader(new InputStreamReader(System.in, StandardCharsets.UTF_8)));
@@ -135,8 +135,8 @@ public static void main(String[] args) {
     final String dbtErrorStack = String.join("\n", errors);
     if (!"".equals(dbtErrorStack)) {
       final Map<ErrorMapKeys, String> errorMap = SentryExceptionHelper.getUsefulErrorMessageAndTypeFromDbtError(dbtErrorStack);
-      String internalMessage = errorMap.get(ErrorMapKeys.ERROR_MAP_MESSAGE_KEY);
-      AirbyteMessage traceMessage = new AirbyteMessage()
+      final String internalMessage = errorMap.get(ErrorMapKeys.ERROR_MAP_MESSAGE_KEY);
+      final AirbyteMessage traceMessage = new AirbyteMessage()
           .withType(Type.TRACE)
           .withTrace(new AirbyteTraceMessage()
               .withType(AirbyteTraceMessage.Type.ERROR)

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.21.1
+version=0.21.2