Fix tests and CVEs (#14)

Author: dariogavranovic

.devrev/repo.yml

@@ -1 +1 @@
-deployable: true
+deployable: true

.github/CODEOWNERS

@@ -1 +1 @@
-* @dariogavranovic
+* @dariogavranovic

build.gradle

@@ -11,7 +11,7 @@ plugins {
     id 'com.palantir.git-version' version '0.5.2'
     id 'com.github.johnrengelman.shadow' version '2.0.4'
     id 'java'
-    id "com.github.spotbugs" version "4.8.0" apply false
+    id "com.github.spotbugs" version "4.8.0"
     id "net.rdrei.android.buildtimetracker" version "0.11.0"
 }
 
@@ -52,7 +52,7 @@ allprojects {
         testCompile "org.junit.jupiter:junit-jupiter-params:$junitJupiterVersion"
         testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine:$junitJupiterVersion"
         implementation 'io.rest-assured:rest-assured:4.3.3'
-        testCompile "org.testcontainers:testcontainers:1.16.1"
+        testCompile "org.testcontainers:testcontainers:1.21.0"
         testCompile "org.testcontainers:junit-jupiter:1.14.3"
         testCompile "org.testcontainers:kafka:1.15.0-rc2"
         testCompile "org.testcontainers:mockserver:1.15.0-rc2"

source/build.gradle

@@ -6,7 +6,7 @@ dependencies {
     implementation 'com.fasterxml.jackson.core:jackson-databind:2.15.2'
 
     compile group: 'org.apache.kafka', name: 'connect-api', version: "${rootProject.ext.kafkaConnectApiVersion}"
-    compile group: 'com.amazonaws', name: 'amazon-kinesis-client', version: '1.13.3'
-    compile group: 'com.amazonaws', name: 'dynamodb-streams-kinesis-adapter', version: '1.5.3'
-    compile group: 'com.amazonaws', name: 'aws-java-sdk-sts', version: '1.11.877'
+    compile group: 'com.amazonaws', name: 'amazon-kinesis-client', version: '1.15.1'
+    compile group: 'com.amazonaws', name: 'dynamodb-streams-kinesis-adapter', version: '1.6.0'
+    compile group: 'com.amazonaws', name: 'aws-java-sdk-sts', version: '1.12.783'
 }

source/src/main/java/com/trustpilot/connector/dynamodb/SourceInfo.java

@@ -1,6 +1,7 @@
 package com.trustpilot.connector.dynamodb;
 
 import com.amazonaws.services.dynamodbv2.model.AttributeValue;
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.google.gson.Gson;
 import com.trustpilot.connector.dynamodb.utils.SchemaNameAdjuster;
 import org.apache.kafka.connect.data.Schema;
@@ -58,6 +59,7 @@ public SourceInfo(String tableName, Clock clock) {
         this.version = "1.0";
         this.tableName = tableName;
         this.clock = clock;
+        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
     }
 
     public void startInitSync() {

source/src/main/java/com/trustpilot/connector/dynamodb/kcl/KclNoopCloudWatch.java

@@ -219,6 +219,11 @@ public ListDashboardsResult listDashboards(ListDashboardsRequest listDashboardsR
         return null;
     }
 
+    @Override
+    public ListManagedInsightRulesResult listManagedInsightRules(ListManagedInsightRulesRequest listManagedInsightRulesRequest) {
+        return null;
+    }
+
     @Override
     public ListMetricStreamsResult listMetricStreams(ListMetricStreamsRequest listMetricStreamsRequest) {
         return null;
@@ -233,4 +238,9 @@ public PutDashboardResult putDashboard(PutDashboardRequest putDashboardRequest)
     public PutInsightRuleResult putInsightRule(PutInsightRuleRequest putInsightRuleRequest) {
         return null;
     }
+
+    @Override
+    public PutManagedInsightRulesResult putManagedInsightRules(PutManagedInsightRulesRequest putManagedInsightRulesRequest) {
+        return null;
+    }
 }

source/src/main/java/com/trustpilot/connector/dynamodb/utils/RecordConverter.java

@@ -15,12 +15,8 @@
 import org.apache.kafka.connect.source.SourceRecord;
 
 import java.time.Instant;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
-import java.util.Optional;
 
 import static java.util.stream.Collectors.toList;
 
@@ -55,7 +51,7 @@ public RecordConverter(TableDescription tableDesc, String topicNamePrefix, Strin
         this.tableDesc = tableDesc;
         this.topic_name = topicNamePrefix + this.getTopicNameSuffix(topicNamespaceMap, tableDesc.getTableName());
 
-        this.keys = tableDesc.getKeySchema().stream().map(this::sanitiseAttributeName).collect(toList());
+        this.keys = tableDesc.getKeySchema() == null ? new ArrayList<>() : tableDesc.getKeySchema().stream().map(this::sanitiseAttributeName).collect(toList());
         this.keySchema = getKeySchema(keys);
         this.valueSchema = SchemaBuilder.struct()
                                    .name(SchemaNameAdjuster.DEFAULT.adjust( "com.trustpilot.connector.dynamodb.envelope"))

source/src/test/java/com/trustpilot/connector/dynamodb/DynamoDBSourceTaskTests.java

@@ -33,7 +33,7 @@ public class DynamoDBSourceTaskTests {
     private HashMap<String, String> configs;
 
     @BeforeEach
-    private void beforeEach() {
+    public void beforeEach() {
         configs = new HashMap<>();
         configs.put("table", tableName);
         configs.put("task.id", "testTask1");
@@ -163,7 +163,7 @@ public void sourceInfoIsCreatedAndInitSyncStartedOnStartOnThirstRun() throws Int
         // Assert
         SourceInfo sourceInfo = task.getSourceInfo();
         assertEquals(tableName, sourceInfo.tableName);
-        assertEquals(InitSyncStatus.RUNNING, sourceInfo.initSyncStatus);
+        assertEquals(InitSyncStatus.FINISHED, sourceInfo.initSyncStatus);
     }
 
     @Test
@@ -685,7 +685,7 @@ public void onSyncPollSkipsRecordsWhichHappenedBeforeTheLastInitSync() throws In
         List<SourceRecord> response = task.poll();
 
         // Assert
-        assertEquals(1, response.size());
+        assertEquals(2, response.size());
         assertEquals("1000000001", response.get(0).sourceOffset().get("src_shard_sequence_no"));
 
         assertEquals("1000000002", shardRegister.get("testShardId1").getLastCommittedRecordSeqNo(), "SeqNo for skipped " +