Guarantee in order delivery for partitions

rockset · Oct 30, 2023 · b140f8e · b140f8e
1 parent 6525cc2
commit b140f8e
Show file tree

Hide file tree

Showing 10 changed files with 149 additions and 488 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,4 +1,9 @@
 # Rockset Kafka Connect Changelog
+## v2.0.0 2023-10-30
+- New configuration option `rockset.retry.backoff.ms`
+- Removed deprecated configurations `rockset.apikey`, `rockset.collection`, and `rockset.workspace`
+- Bug fix for potential out-of-order message delivery
+
 ## v1.4.3 2023-09-15
 - Update rockset-java client dependency
 

diff --git a/pom.xml b/pom.xml
@@ -5,7 +5,7 @@
 
     <groupId>rockset</groupId>
     <artifactId>kafka-connect-rockset</artifactId>
-    <version>1.4.3</version>
+    <version>2.0.0</version>
     <packaging>jar</packaging>
 
     <name>kafka-connect-rockset</name>
@@ -38,6 +38,12 @@
             <artifactId>kafka-connect-avro-converter</artifactId>
             <version>7.2.2</version>
         </dependency>
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>6.14.2</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

diff --git a/src/main/java/rockset/RocksetConnectorConfig.java b/src/main/java/rockset/RocksetConnectorConfig.java
@@ -14,15 +14,13 @@
 import org.slf4j.LoggerFactory;
 
 public class RocksetConnectorConfig extends AbstractConfig {
-  private static Logger log = LoggerFactory.getLogger(RocksetConnectorConfig.class);
+  private static final Logger log = LoggerFactory.getLogger(RocksetConnectorConfig.class);
   public static final String FORMAT = "format";
   public static final String ROCKSET_APISERVER_URL = "rockset.apiserver.url";
-  public static final String ROCKSET_APIKEY = "rockset.apikey";
   public static final String ROCKSET_INTEGRATION_KEY = "rockset.integration.key";
-  public static final String ROCKSET_COLLECTION = "rockset.collection";
-  public static final String ROCKSET_WORKSPACE = "rockset.workspace";
   public static final String ROCKSET_TASK_THREADS = "rockset.task.threads";
   public static final String ROCKSET_BATCH_SIZE = "rockset.batch.size";
+  public static final String ROCKSET_RETRY_BACKOFF_MS = "rockset.retry.backoff.ms";
 
   private RocksetConnectorConfig(ConfigDef config, Map<String, String> originals) {
     super(config, originals, true);
@@ -44,7 +42,7 @@ public static ConfigDef config() {
                 .documentation("Rockset API Server URL")
                 .importance(Importance.HIGH)
                 .validator(RocksetConnectorConfig::validateApiServer)
-                .defaultValue("https://api.rs2.usw2.rockset.com")
+                .defaultValue("https://api.usw2a1.rockset.com")
                 .build())
         .define(
             ConfigKeyBuilder.of(ROCKSET_INTEGRATION_KEY, Type.STRING)
@@ -73,24 +71,10 @@ public static ConfigDef config() {
                 .defaultValue("json")
                 .build())
         .define(
-            ConfigKeyBuilder.of(ROCKSET_APIKEY, Type.STRING)
-                .documentation("(Deprecated) Rockset API Key")
-                .importance(Importance.HIGH)
-                .defaultValue(null)
-                .build())
-        .define(
-            ConfigKeyBuilder.of(ROCKSET_COLLECTION, Type.STRING)
-                .documentation(
-                    "(Deprecated) Rockset collection that incoming documents will be written to.")
-                .importance(Importance.HIGH)
-                .defaultValue(null)
-                .build())
-        .define(
-            ConfigKeyBuilder.of(ROCKSET_WORKSPACE, Type.STRING)
-                .documentation(
-                    "(Deprecated) Rockset workspace that incoming documents will be written to.")
-                .importance(Importance.HIGH)
-                .defaultValue("commons")
+            ConfigKeyBuilder.of(ROCKSET_RETRY_BACKOFF_MS, Type.INT)
+                .documentation("How long to backoff in milliseconds between retriable errors")
+                .importance(Importance.MEDIUM)
+                .defaultValue(5000)
                 .build());
   }
 
@@ -154,4 +138,8 @@ public int getRocksetBatchSize() {
   public String getFormat() {
     return this.getString(FORMAT);
   }
+
+  public int getRetryBackoffMs() {
+    return this.getInt(ROCKSET_RETRY_BACKOFF_MS);
+  }
 }
diff --git a/src/main/java/rockset/RocksetRequestWrapper.java b/src/main/java/rockset/RocksetRequestWrapper.java
@@ -31,22 +31,16 @@ public class RocksetRequestWrapper implements RequestWrapper {
   private static final String KAFKA_ENDPOINT = "/v1/receivers/kafka";
   private static final ObjectMapper mapper = new ObjectMapper();
 
-  private OkHttpClient client;
-  private String integrationKeyEncoded;
-  private String apiServer;
+  private final OkHttpClient client;
+  private final String integrationKeyEncoded;
+  private final String apiServer;
 
-  // used for testing
   public RocksetRequestWrapper(RocksetConnectorConfig config, OkHttpClient client) {
     this.client = client;
-
-    parseConnectionString(config.getRocksetIntegrationKey());
+    this.integrationKeyEncoded = base64EncodeAsUserPassword(config.getRocksetIntegrationKey());
     this.apiServer = config.getRocksetApiServerUrl();
   }
 
-  private void parseConnectionString(String integrationKey) {
-    this.integrationKeyEncoded = base64EncodeAsUserPassword(integrationKey);
-  }
-
   private static String base64EncodeAsUserPassword(String integrationKey) {
     final String userPassword = integrationKey + ":"; // password is empty
     return Base64.getEncoder().encodeToString(userPassword.getBytes(StandardCharsets.UTF_8));
@@ -93,6 +87,12 @@ private boolean isSuccessHttpCode(int code) {
 
   private void sendDocs(String topic, List<KafkaMessage> messages) {
     Preconditions.checkArgument(!messages.isEmpty());
+    if (Thread.interrupted()) {
+      // Exit early in case another thread failed and the whole batch of requests from
+      // RocksetSinkTask::put()
+      // are going to be cancelled
+      throw new ConnectException("Interrupted during call to RocksetRequestWrapper::addDocs");
+    }
     log.debug("Sending batch of {} messages for topic: {} to Rockset", messages.size(), topic);
 
     KafkaDocumentsRequest documentsRequest =