Source-DynamoDB : Fix failing build (#22682)

* Fix failing DynamoDB source build * Bump version
airbytehq · Feb 10, 2023 · 762a8cd · 762a8cd
1 parent 6058e42
commit 762a8cd
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 7 deletions.
diff --git a/airbyte-integrations/connectors/source-dynamodb/Dockerfile b/airbyte-integrations/connectors/source-dynamodb/Dockerfile
@@ -17,5 +17,5 @@ ENV APPLICATION source-dynamodb
 COPY --from=build /airbyte /airbyte
 
 # Airbyte's build system uses these labels to know what to name and tag the docker images produced by this Dockerfile.
-LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.version=0.1.1
 LABEL io.airbyte.name=airbyte/source-dynamodb
diff --git a/.../source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbUtils.java b/.../source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbUtils.java
@@ -26,8 +26,8 @@ private DynamodbUtils() {
 
   }
 
-  public static DynamoDbClient createDynamoDbClient(DynamodbConfig dynamodbConfig) {
-    var dynamoDbClientBuilder = DynamoDbClient.builder();
+  public static DynamoDbClient createDynamoDbClient(final DynamodbConfig dynamodbConfig) {
+    final var dynamoDbClientBuilder = DynamoDbClient.builder();
 
     // configure access credentials
     dynamoDbClientBuilder.credentialsProvider(StaticCredentialsProvider.create(
@@ -44,7 +44,7 @@ public static DynamoDbClient createDynamoDbClient(DynamodbConfig dynamodbConfig)
     return dynamoDbClientBuilder.build();
   }
 
-  public static AirbyteMessage mapAirbyteMessage(String stream, JsonNode data) {
+  public static AirbyteMessage mapAirbyteMessage(final String stream, final JsonNode data) {
     return new AirbyteMessage()
         .withType(AirbyteMessage.Type.RECORD)
         .withRecord(new AirbyteRecordMessage()
@@ -53,12 +53,13 @@ public static AirbyteMessage mapAirbyteMessage(String stream, JsonNode data) {
             .withData(data));
   }
 
-  public static StreamState deserializeStreamState(JsonNode state, boolean useStreamCapableState) {
-    Optional<StateWrapper> typedState =
+  public static StreamState deserializeStreamState(final JsonNode state, final boolean useStreamCapableState) {
+    final Optional<StateWrapper> typedState =
         StateMessageHelper.getTypedState(state, useStreamCapableState);
     return typedState.map(stateWrapper -> switch (stateWrapper.getStateType()) {
       case STREAM:
-        yield new StreamState(AirbyteStateMessage.AirbyteStateType.STREAM, stateWrapper.getStateMessages());
+        yield new StreamState(AirbyteStateMessage.AirbyteStateType.STREAM,
+            stateWrapper.getStateMessages().stream().map(DynamodbUtils::convertStateMessage).toList());
       case LEGACY:
         yield new StreamState(AirbyteStateMessage.AirbyteStateType.LEGACY, List.of(
             new AirbyteStateMessage().withType(AirbyteStateMessage.AirbyteStateType.LEGACY)
@@ -79,6 +80,10 @@ yield new StreamState(AirbyteStateMessage.AirbyteStateType.LEGACY, List.of(
     });
   }
 
+  private static AirbyteStateMessage convertStateMessage(final io.airbyte.protocol.models.AirbyteStateMessage state) {
+    return Jsons.object(Jsons.jsonNode(state), AirbyteStateMessage.class);
+  }
+
   record StreamState(
 
                      AirbyteStateMessage.AirbyteStateType airbyteStateType,

diff --git a/docs/integrations/sources/dynamodb.md b/docs/integrations/sources/dynamodb.md
@@ -61,4 +61,5 @@ This guide describes in details how you can configure the connector to connect w
 
 | Version | Date       | Pull Request | Subject         |
 |:--------|:-----------|:-------------|:----------------|
+| 0.1.1   | 02-09-2023 | https://github.com/airbytehq/airbyte/pull/22682             | Build fixes     |
 | 0.1.0   | 11-14-2022 | https://github.com/airbytehq/airbyte/pull/18750             | Initial version |