Remove unimplemented Kinesis config

trinodb · Jan 13, 2022 · 4d15192 · 4d15192
1 parent 5170f8b
commit 4d15192
Show file tree

Hide file tree

Showing 4 changed files with 2 additions and 22 deletions.
diff --git a/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisConfig.java b/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisConfig.java
@@ -15,6 +15,7 @@
 
 import io.airlift.configuration.Config;
 import io.airlift.configuration.ConfigDescription;
+import io.airlift.configuration.DefunctConfig;
 import io.airlift.units.Duration;
 import io.airlift.units.MinDuration;
 
@@ -24,6 +25,7 @@
 
 import java.util.concurrent.TimeUnit;
 
+@DefunctConfig("kinesis.checkpoint-interval")
 public class KinesisConfig
 {
     private String defaultSchema = "default";
@@ -43,7 +45,6 @@ public class KinesisConfig
     private boolean checkpointEnabled;
     private long dynamoReadCapacity = 50L;
     private long dynamoWriteCapacity = 10L;
-    private Duration checkpointInterval = new Duration(60000, TimeUnit.MILLISECONDS);
     private String logicalProcessName = "process1";
     private int iteratorNumber;
 
@@ -277,19 +278,6 @@ public KinesisConfig setDynamoWriteCapacity(long dynamoWriteCapacity)
         return this;
     }
 
-    public Duration getCheckpointInterval()
-    {
-        return checkpointInterval;
-    }
-
-    @Config("kinesis.checkpoint-interval")
-    @ConfigDescription("Intervals at which to checkpoint shard iterator details")
-    public KinesisConfig setCheckpointInterval(Duration checkpointInterval)
-    {
-        this.checkpointInterval = checkpointInterval;
-        return this;
-    }
-
     public String getLogicalProcessName()
     {
         return logicalProcessName;

diff --git a/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java b/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisRecordSet.java
@@ -115,7 +115,6 @@ public class KinesisRecordSet
         requireNonNull(kinesisConfig, "kinesisConfig is null");
         long dynamoReadCapacity = kinesisConfig.getDynamoReadCapacity();
         long dynamoWriteCapacity = kinesisConfig.getDynamoWriteCapacity();
-        long checkPointIntervalMillis = kinesisConfig.getCheckpointInterval().toMillis();
         this.isLogBatches = kinesisConfig.isLogBatches();
 
         this.clientManager = requireNonNull(clientManager, "clientManager is null");
@@ -159,7 +158,6 @@ public class KinesisRecordSet
                     split,
                     logicalProcessName,
                     curIterationNumber,
-                    checkPointIntervalMillis,
                     dynamoReadCapacity,
                     dynamoWriteCapacity);
 

diff --git a/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisShardCheckpointer.java b/plugin/trino-kinesis/src/main/java/io/trino/plugin/kinesis/KinesisShardCheckpointer.java
@@ -37,15 +37,13 @@ public KinesisShardCheckpointer(
             KinesisSplit kinesisSplit,
             String logicalProcessName,
             int currentIterationNumber,
-            long checkpointIntervalMS,
             long dynamoReadCapacity,
             long dynamoWriteCapacity)
     {
         this(new KinesisClientLeaseManager(dynamoDBTable, dynamoDBClient),
                 kinesisSplit,
                 logicalProcessName,
                 currentIterationNumber,
-                checkpointIntervalMS,
                 dynamoReadCapacity,
                 dynamoWriteCapacity);
     }
@@ -55,7 +53,6 @@ public KinesisShardCheckpointer(
             KinesisSplit kinesisSplit,
             String logicalProcessName,
             int currentIterationNumber,
-            long checkpointIntervalMS,
             long dynamoReadCapacity,
             long dynamoWriteCapacity)
     {

diff --git a/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestKinesisConfig.java b/plugin/trino-kinesis/src/test/java/io/trino/plugin/kinesis/TestKinesisConfig.java
@@ -44,7 +44,6 @@ public void testDefaults()
                 .setCheckpointEnabled(false)
                 .setDynamoReadCapacity(50)
                 .setDynamoWriteCapacity(10)
-                .setCheckpointInterval(new Duration(60000, TimeUnit.MILLISECONDS))
                 .setLogicalProcessName("process1")
                 .setIteratorNumber(0));
     }
@@ -70,7 +69,6 @@ public void testExplicitPropertyMappings()
                 .put("kinesis.checkpoint-enabled", "true")
                 .put("kinesis.dynamo-read-capacity", "100")
                 .put("kinesis.dynamo-write-capacity", "20")
-                .put("kinesis.checkpoint-interval", "50000ms")
                 .put("kinesis.checkpoint-logical-name", "process")
                 .put("kinesis.iterator-number", "1")
                 .build();
@@ -93,7 +91,6 @@ public void testExplicitPropertyMappings()
                 .setCheckpointEnabled(true)
                 .setDynamoReadCapacity(100)
                 .setDynamoWriteCapacity(20)
-                .setCheckpointInterval(new Duration(50000, TimeUnit.MILLISECONDS))
                 .setLogicalProcessName("process")
                 .setIteratorNumber(1);