Merge branch 'master' into fleetio-source-connector-cursor-pagination

.github/workflows/connector_teams_review_requirements.yml

@@ -9,9 +9,11 @@ on:
       - synchronize
     paths:
       - "airbyte-integrations/connectors/source-**"
+      - "airbyte-integrations/connectors/destination-**"
   pull_request_review:
     paths:
       - "airbyte-integrations/connectors/source-**"
+      - "airbyte-integrations/connectors/destination-**"
 jobs:
   check-review-requirements:
     name: "Check if a review is required from Connector teams"

.github/workflows/gradle.yml

@@ -45,13 +45,19 @@ jobs:
         uses: tj-actions/changed-files@v39
         id: changes
         with:
+          # Include java connectors and java CDK.
+          # Adding all *.java and *.gradle files gets us most of the way there.
+          # We're take a bit more strict for the java CDK, to make sure that
+          # the tests run when they should, for instance when changing the contents
+          # of a text file used as a resource.
           files_yaml: |
             gradlecheck:
-              - '**/*'
-              - '!**/*.md'
-              - '!.github/*'
+              - '**/*.java'
+              - '**/*.gradle'
+              - 'airbyte-cdk/java/**/*'
 
       - uses: actions/setup-java@v3
+        if: steps.changes.outputs.gradlecheck_any_changed == 'true'
         with:
           distribution: "zulu"
           java-version: "21"
@@ -62,6 +68,7 @@ jobs:
         if: steps.changes.outputs.gradlecheck_any_changed == 'true'
         run: python3 -m pip install virtualenv --user
       - name: Docker login
+        if: steps.changes.outputs.gradlecheck_any_changed == 'true'
         # Some tests use testcontainers which pull images from DockerHub.
         uses: docker/login-action@v1
         with:

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

@@ -166,6 +166,8 @@ MavenLocal debugging steps:
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.21.3  | 2024-02-20 | [\#35394](https://github.com/airbytehq/airbyte/pull/35394) | Add Junit progress information to the test logs                                                                                                                                  |
+| 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.3

airbyte-cdk/java/airbyte-cdk/core/src/test/java/io/airbyte/cdk/integrations/base/AirbyteLogMessageTemplateTest.java

@@ -25,59 +25,54 @@
 import org.apache.logging.log4j.core.config.Configuration;
 import org.apache.logging.log4j.core.config.Configurator;
 import org.apache.logging.log4j.core.config.LoggerConfig;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
+import org.apache.logging.log4j.spi.ExtendedLogger;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.junit.platform.commons.util.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class AirbyteLogMessageTemplateTest {
 
-  private static final ByteArrayOutputStream outputContent = new ByteArrayOutputStream();
-  private static final Logger LOGGER = LoggerFactory.getLogger(AirbyteLogMessageTemplateTest.class);
   public static final String OUTPUT_STREAM_APPENDER = "OutputStreamAppender";
   public static final String CONSOLE_JSON_APPENDER = "ConsoleJSONAppender";
-  private static OutputStreamAppender outputStreamAppender;
-  private static LoggerConfig rootLoggerConfig;
-  private static LoggerContext loggerContext;
+  private LoggerContext loggerContext;
+  private LoggerConfig rootLoggerConfig;
+  private ExtendedLogger logger;
+  private OutputStreamAppender outputStreamAppender;
+  private ByteArrayOutputStream outputContent;
 
-  @BeforeAll
-  static void init() {
+  void getLogger() {
     // We are creating a log appender with the same output pattern
     // as the console json appender defined in this project's log4j2.xml file.
     // We then attach this log appender with the LOGGER instance so that we can validate the logs
     // produced by code and assert that it matches the expected format.
     loggerContext = Configurator.initialize(null, "log4j2.xml");
+
     final Configuration configuration = loggerContext.getConfiguration();
     rootLoggerConfig = configuration.getLoggerConfig("");
 
+    outputContent = new ByteArrayOutputStream();
     outputStreamAppender = OutputStreamAppender.createAppender(
         rootLoggerConfig.getAppenders().get(CONSOLE_JSON_APPENDER).getLayout(),
         null, outputContent, OUTPUT_STREAM_APPENDER, false, true);
     outputStreamAppender.start();
 
     rootLoggerConfig.addAppender(outputStreamAppender, Level.ALL, null);
+    logger = loggerContext.getLogger(AirbyteLogMessageTemplateTest.class);
   }
 
-  @BeforeEach
-  void setup() {
-    outputContent.reset();
-  }
-
-  @AfterAll
-  static void cleanUp() {
+  @AfterEach
+  void closeLogger() {
     outputStreamAppender.stop();
     rootLoggerConfig.removeAppender(OUTPUT_STREAM_APPENDER);
     loggerContext.close();
   }
 
   @Test
   public void testAirbyteLogMessageFormat() throws java.io.IOException {
-    LOGGER.info("hello");
+    getLogger();
+    logger.info("hello");
 
     outputContent.flush();
     final String logMessage = outputContent.toString(StandardCharsets.UTF_8);
@@ -114,12 +109,13 @@ private AirbyteLogMessage validateAirbyteMessageIsLog(final AirbyteMessage airby
 
   @ParameterizedTest
   @ValueSource(ints = {2, 100, 9000})
-  public void testAirbyteLogMessageLength(int stringRepeatitions) throws java.io.IOException {
+  public void testAirbyteLogMessageLength(int stringRepetitions) throws java.io.IOException {
+    getLogger();
     final StringBuilder sb = new StringBuilder();
-    for (int i = 0; i < stringRepeatitions; i++) {
+    for (int i = 0; i < stringRepetitions; i++) {
       sb.append("abcd");
     }
-    LOGGER.info(sb.toString(), new RuntimeException("aaaaa bbbbbb ccccccc dddddd"));
+    logger.info(sb.toString(), new RuntimeException("aaaaa bbbbbb ccccccc dddddd"));
     outputContent.flush();
     final String logMessage = outputContent.toString(StandardCharsets.UTF_8);