bubble up debezium config error messages as exceptions

airbytehq · Dec 19, 2023 · 86a9add · 86a9add
1 parent 26d3746
commit 86a9add
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 7 deletions.
diff --git a/airbyte-cdk/java/airbyte-cdk/README.md b/airbyte-cdk/java/airbyte-cdk/README.md
@@ -164,6 +164,7 @@ MavenLocal debugging steps:
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.8.1   | 2023-12-19 | [\#33658](https://github.com/airbytehq/airbyte/pull/33658) | Always fail when debezium fails, even if it happened during the setup phase.                                                                                   |
 | 0.8.0   | 2023-12-18 | [\#33506](https://github.com/airbytehq/airbyte/pull/33506) | Improve async destination shutdown logic; more JDBC async migration work; improve DAT test schema handling                                                     |
 | 0.7.9   | 2023-12-18 | [\#33549](https://github.com/airbytehq/airbyte/pull/33549) | Improve MongoDB logging.                                                                                                                                       |
 | 0.7.8   | 2023-12-18 | [\#33365](https://github.com/airbytehq/airbyte/pull/33365) | Emit stream statuses more consistently                                                                                                                         |

diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties b/airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties
@@ -1 +1 @@
-version=0.8.0
+version=0.8.1
diff --git a/...src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumRecordPublisher.java b/...src/main/java/io/airbyte/cdk/integrations/debezium/internals/DebeziumRecordPublisher.java
@@ -86,12 +86,14 @@ public void start(final BlockingQueue<ChangeEvent<String, String>> queue) {
         .using((success, message, error) -> {
           LOGGER.info("Debezium engine shutdown. Engine terminated successfully : {}", success);
           LOGGER.info(message);
-          thrownError.set(error);
-          // If debezium has not shutdown correctly, it can indicate an error with the connector configuration
-          // or a partial sync success.
-          // In situations like these, the preference is to fail loud and clear.
-          if (thrownError.get() != null && !success) {
-            thrownError.set(new RuntimeException(message));
+          if (!success) {
+            if (error != null) {
+              thrownError.set(error);
+            } else {
+              // There are cases where Debezium doesn't succeed but only fills the message field.
+              // In that case, we still want to fail loud and clear
+              thrownError.set(new RuntimeException(message));
+            }
           }
           engineLatch.countDown();
         })