Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Un marshall dynamo documents #4

Merged
merged 2 commits into from
Sep 12, 2022
Merged

Un marshall dynamo documents #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4d8f24a
unMarshalling Dynamo json records
gurjit-sandhu Sep 11, 2022
90e852b
updated test cases
gurjit-sandhu Sep 12, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion source/src/main/java/com/trustpilot/connector/dynamodb/utils/RecordConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.TableDescription;
import com.amazonaws.services.dynamodbv2.document.ItemUtils;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Strings;
Expand Down Expand Up @@ -74,6 +75,9 @@ public SourceRecord toSourceRecord(
LinkedHashMap::new
));

// getUnmarshallItems from Dynamo Document
Map<String, Object> unMarshalledItems = ItemUtils.toSimpleMapValue(attributes);

// Leveraging offsets to store shard and sequence number with each item pushed to Kafka.
// This info will only be used to update `shardRegister` and won't be used to reset state after restart
Map<String, Object> offsets = SourceInfo.toOffset(sourceInfo);
Expand Down Expand Up @@ -101,7 +105,7 @@ public SourceRecord toSourceRecord(

Struct valueData = new Struct(valueSchema)
.put(Envelope.FieldName.VERSION, sourceInfo.version)
.put(Envelope.FieldName.DOCUMENT, objectMapper.writeValueAsString(sanitisedAttributes))
.put(Envelope.FieldName.DOCUMENT, objectMapper.writeValueAsString(unMarshalledItems))
.put(Envelope.FieldName.SOURCE, SourceInfo.toStruct(sourceInfo))
.put(Envelope.FieldName.OPERATION, op.code())
.put(Envelope.FieldName.TIMESTAMP, arrivalTimestamp.toEpochMilli());
Expand Down
6 changes: 3 additions & 3 deletions source/src/test/java/com/trustpilot/connector/dynamodb/DynamoDBSourceTaskTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ public void onInitSyncRunningPollReturnsScannedItemsBatch() throws InterruptedEx

assertEquals(1, response.size());
assertEquals("r", ((Struct) response.get(0).value()).getString("op"));
assertEquals("{\"col2\":{\"s\":\"val1\"},\"col3\":{\"n\":\"1\"},\"col1\":{\"s\":\"key1\"}}", ((Struct) response.get(0).value()).getString("document"));
assertEquals(({"col2":"val1","col3":1,"col1":"key1"}), ((Struct) response.get(0).value()).getString("document"));
assertEquals(InitSyncStatus.RUNNING, task.getSourceInfo().initSyncStatus);
assertEquals(exclusiveStartKey, task.getSourceInfo().exclusiveStartKey);
}
Expand Down Expand Up @@ -560,8 +560,8 @@ public void onSyncPollReturnsReceivedRecords() throws InterruptedException {

// Assert
assertEquals(3, response.size());
assertEquals("{\"col2\":{\"s\":\"val1\"},\"col3\":{\"n\":\"1\"},\"col1\":{\"s\":\"key1\"}}", ((Struct) response.get(0).value()).getString("document"));
assertEquals("{\"col1\":{\"s\":\"key2\"}}", ((Struct) response.get(1).value()).getString("document"));
assertEquals("{\"col2\":\"val1\",\"col3\":1,\"col1\":\"key1\"}", ((Struct) response.get(0).value()).getString("document"));
assertEquals("{\"col1\":\"key2\"}", ((Struct) response.get(1).value()).getString("document"));
assertNull(response.get(2).value()); // tombstone
}

Expand Down
4 changes: 2 additions & 2 deletions source/src/test/java/com/trustpilot/connector/dynamodb/utils/RecordConverterTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,7 @@ public void recordAttributesAreAddedToValueData() throws Exception {
);

// Assert
assertEquals("{\"testKV1\":{\"s\":\"testKV1Value\"},\"testKV2\":{\"s\":\"2\"},\"testV2\":{\"s\":\"testStringValue\"},\"testV1\":{\"n\":\"1\"}}",
assertEquals("{\"testKV1\":\"testKV1Value\",\"testKV2\":\"2\",\"testV2\":\"testStringValue\",\"testV1\":1}",
((Struct) record.value()).getString("document"));
}

Expand Down Expand Up @@ -271,7 +271,7 @@ public void recordAttributesAreAddedToValueDataWhenAttributesContainsInvalidChar
"testSequenceNumberID1"
);

String expected = "{\"test1234\":{\"s\":\"testKV1Value\"},\"_starts_with_underscore\":{\"n\":\"1\"},\"startswithnumber\":{\"s\":\"2\"},\"test\":{\"s\":\"testStringValue\"}}";
String expected = "{\"test-1234\":\"testKV1Value\",\"_starts_with_underscore\":1,\"1-starts-with-number\":\"2\",\"test!@£$%^\":\"testStringValue\"}";

// Assert
assertEquals(expected,
Expand Down