jprdonnelly/airbyte
1
0
Fork 0
mirror of synced 2025-12-25 02:09:19 -05:00
Code Issues Packages Projects Releases Wiki Activity

jdbc-sources: migrate to PER-STREAM state (#33485)

Browse Source
This commit is contained in:
Subodh Kant Chaturvedi
2023-12-19 15:38:39 +05:30
committed by GitHub
parent ff34abd641
commit 0c3d12ba2b
91 changed files with 1365 additions and 875 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
airbyte-integrations/connectors/source-dynamodb/acceptance-test-config.yml
View File

@@ -1,7 +0,0 @@
# See [Connector Acceptance Tests](https://docs.airbyte.com/connector-development/testing-connectors/connector-acceptance-tests-reference)
# for more information about how to configure these tests
connector_image: airbyte/source-dynamodb:dev
acceptance-tests:
spec:
tests:
- spec_path: "main/resources/spec.json"

4
airbyte-integrations/connectors/source-dynamodb/build.gradle
View File

@@ -4,7 +4,7 @@ plugins {
}
airbyteJavaConnector {
cdkVersionRequired = '0.2.0'
cdkVersionRequired = '0.7.7'
features = ['db-sources']
useLocalCdk = false
}
@@ -42,4 +42,6 @@ dependencies {
testImplementation "org.assertj:assertj-core:${assertVersion}"
testImplementation "org.testcontainers:localstack:${testContainersVersion}"
integrationTestJavaImplementation 'com.amazonaws:aws-java-sdk:1.12.610'
}

2
airbyte-integrations/connectors/source-dynamodb/metadata.yaml
View File

@@ -5,7 +5,7 @@ data:
connectorSubtype: api
connectorType: source
definitionId: 50401137-8871-4c5a-abb7-1f5fda35545a
dockerImageTag: 0.1.2
dockerImageTag: 0.2.0
dockerRepository: airbyte/source-dynamodb
documentationUrl: https://docs.airbyte.com/integrations/sources/dynamodb
githubIssueLabel: source-dynamodb

6
airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbSource.java
View File

@@ -16,8 +16,6 @@ import io.airbyte.cdk.integrations.source.relationaldb.CursorInfo;
import io.airbyte.cdk.integrations.source.relationaldb.StateDecoratingIterator;
import io.airbyte.cdk.integrations.source.relationaldb.state.StateManager;
import io.airbyte.cdk.integrations.source.relationaldb.state.StateManagerFactory;
import io.airbyte.commons.features.EnvVariableFeatureFlags;
import io.airbyte.commons.features.FeatureFlags;
import io.airbyte.commons.json.Jsons;
import io.airbyte.commons.stream.AirbyteStreamUtils;
import io.airbyte.commons.util.AutoCloseableIterator;
@@ -42,8 +40,6 @@ public class DynamodbSource extends BaseConnector implements Source {
private static final Logger LOGGER = LoggerFactory.getLogger(DynamodbSource.class);
private final FeatureFlags featureFlags = new EnvVariableFeatureFlags();
private final ObjectMapper objectMapper = new ObjectMapper();
public static void main(final String[] args) throws Exception {
@@ -98,7 +94,7 @@ public class DynamodbSource extends BaseConnector implements Source {
final ConfiguredAirbyteCatalog catalog,
final JsonNode state) {
final var streamState = DynamodbUtils.deserializeStreamState(state, featureFlags.useStreamCapableState());
final var streamState = DynamodbUtils.deserializeStreamState(state);
final StateManager stateManager = StateManagerFactory
.createStateManager(streamState.airbyteStateType(), streamState.airbyteStateMessages(), catalog);

18
airbyte-integrations/connectors/source-dynamodb/src/main/java/io/airbyte/integrations/source/dynamodb/DynamodbUtils.java
View File

@@ -5,7 +5,6 @@
package io.airbyte.integrations.source.dynamodb;
import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.cdk.integrations.source.relationaldb.models.DbState;
import io.airbyte.commons.json.Jsons;
import io.airbyte.configoss.StateWrapper;
import io.airbyte.configoss.helpers.StateMessageHelper;
@@ -53,9 +52,9 @@ public class DynamodbUtils {
.withData(data));
}
public static StreamState deserializeStreamState(final JsonNode state, final boolean useStreamCapableState) {
public static StreamState deserializeStreamState(final JsonNode state) {
final Optional<StateWrapper> typedState =
StateMessageHelper.getTypedState(state, useStreamCapableState);
StateMessageHelper.getTypedState(state);
return typedState.map(stateWrapper -> switch (stateWrapper.getStateType()) {
case STREAM:
yield new StreamState(AirbyteStateMessage.AirbyteStateType.STREAM,
@@ -68,15 +67,10 @@ public class DynamodbUtils {
throw new UnsupportedOperationException("Unsupported stream state");
}).orElseGet(() -> {
// create empty initial state
if (useStreamCapableState) {
return new StreamState(AirbyteStateMessage.AirbyteStateType.STREAM, List.of(
new AirbyteStateMessage().withType(AirbyteStateMessage.AirbyteStateType.STREAM)
.withStream(new AirbyteStreamState())));
} else {
return new StreamState(AirbyteStateMessage.AirbyteStateType.LEGACY, List.of(
new AirbyteStateMessage().withType(AirbyteStateMessage.AirbyteStateType.LEGACY)
.withData(Jsons.jsonNode(new DbState()))));
}
return new StreamState(AirbyteStateMessage.AirbyteStateType.STREAM, List.of(
new AirbyteStateMessage().withType(AirbyteStateMessage.AirbyteStateType.STREAM)
.withStream(new AirbyteStreamState())));
});
}

2
airbyte-integrations/connectors/source-dynamodb/src/test-integration/java/io/airbyte/integrations/source/dynamodb/DynamodbOperationsTest.java
View File

@@ -15,12 +15,14 @@ import java.util.Set;
import org.json.JSONException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.skyscreamer.jsonassert.JSONAssert;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
@Disabled
public class DynamodbOperationsTest {
private static final String TABLE_NAME = "airbyte_table";

2
airbyte-integrations/connectors/source-dynamodb/src/test-integration/java/io/airbyte/integrations/source/dynamodb/DynamodbSourceAcceptanceTest.java
View File

@@ -13,10 +13,12 @@ import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog;
import io.airbyte.protocol.models.v0.ConnectorSpecification;
import java.util.HashMap;
import java.util.Map;
import org.junit.jupiter.api.Disabled;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
@Disabled
public class DynamodbSourceAcceptanceTest extends SourceAcceptanceTest {
private static final String TABLE_NAME = "airbyte_table";

2
airbyte-integrations/connectors/source-dynamodb/src/test-integration/java/io/airbyte/integrations/source/dynamodb/DynamodbSourceTest.java
View File

@@ -17,11 +17,13 @@ import java.util.Map;
import java.util.stream.Stream;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
@Disabled
public class DynamodbSourceTest {
private static final String TABLE_NAME = "airbyte_table";