SAMZA-2657: Blob Store as backend for Samza State backup and restore

build.gradle

@@ -191,6 +191,7 @@ project(":samza-core_$scalaSuffix") {
     compile "org.apache.commons:commons-lang3:$commonsLang3Version"
     compile "commons-io:commons-io:$commonsIoVersion"
     compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
+    compile "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$jacksonVersion"
     compile "org.eclipse.jetty:jetty-webapp:$jettyVersion"
     compile "org.scala-lang:scala-library:$scalaVersion"
     compile "org.slf4j:slf4j-api:$slf4jVersion"

samza-api/src/main/java/org/apache/samza/storage/BlobStoreAdminFactory.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.storage;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.job.model.JobModel;
+
+
+/**
+ * Factory to create instance of {@link StateBackendAdmin}s that needs to be implemented for every
+ * state backend
+ */
+public interface BlobStoreAdminFactory {
+  /**
+   * Returns an instance of {@link StateBackendAdmin}
+   * @param config job configuration
+   * @param jobModel Job Model
+   */
+  StateBackendAdmin getStateBackendAdmin(Config config, JobModel jobModel);
+}

samza-api/src/main/java/org/apache/samza/storage/TaskStorageAdmin.java → samza-api/src/main/java/org/apache/samza/storage/StateBackendAdmin.java

@@ -20,11 +20,16 @@
 package org.apache.samza.storage;
 
 /**
- * Creates and validate resources for the StateBackendFactory
+ * Admin responsible for loading any resources related to state backend
  */
-public interface TaskStorageAdmin {
-
+public interface StateBackendAdmin {
+  /**
+   * Create all the resources required per job per store state backend
+   */
   void createResources();
 
+  /**
+   * Validate all resources required per job per state for state backend
+   */
   void validateResources();
 }

samza-api/src/main/java/org/apache/samza/storage/StateBackendFactory.java

@@ -24,18 +24,20 @@
 import org.apache.samza.config.Config;
 import org.apache.samza.context.ContainerContext;
 import org.apache.samza.context.JobContext;
+import org.apache.samza.job.model.ContainerModel;
+import org.apache.samza.job.model.JobModel;
 import org.apache.samza.job.model.TaskModel;
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.util.Clock;
 
 
 /**
- * Factory to build the Samza {@link TaskBackupManager}, {@link TaskRestoreManager} and {@link TaskStorageAdmin}
+ * Factory to build the Samza {@link TaskBackupManager}, {@link TaskRestoreManager} and {@link StateBackendAdmin}
  * for a particular state storage backend, which are used to durably backup the Samza task state.
  */
 public interface StateBackendFactory {
   TaskBackupManager getBackupManager(JobContext jobContext,
-      ContainerContext containerContext,
+      ContainerModel containerModel,
       TaskModel taskModel,
       ExecutorService backupExecutor,
       MetricsRegistry taskInstanceMetricsRegistry,
@@ -55,5 +57,5 @@ TaskRestoreManager getRestoreManager(JobContext jobContext,
       File nonLoggedStoreBaseDir,
       KafkaChangelogRestoreParams kafkaChangelogRestoreParams);
 
-  TaskStorageAdmin getAdmin();
+  StateBackendAdmin getAdmin(JobModel jobModel, Config config);
 }

samza-api/src/main/java/org/apache/samza/storage/blobstore/BlobStoreManager.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.storage.blobstore;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.concurrent.CompletionStage;
+import org.apache.samza.annotation.InterfaceStability;
+
+
+/**
+ * Provides interface for common blob store operations: GET, PUT and DELETE
+ */
+@InterfaceStability.Unstable
+public interface BlobStoreManager {
+  /**
+   * Initialize underlying blob store client, if necessary.
+   *
+   */
+  void init();
+
+  /**
+   * Non-blocking PUT call to remote blob store with supplied metadata
+   * @param inputStream InputStream to read the file
+   * @param metadata user supplied {@link Metadata} of the request
+   * @return a future containing the blob ID of the uploaded blob if the upload is successful.
+   */
+  CompletionStage<String> put(InputStream inputStream, Metadata metadata);
+
+  /**
+   * Non-blocking GET call to remote blob store
+   * @param id Blob ID of the blob to get
+   * @param outputStream OutputStream to write the downloaded blob
+   * @param metadata User supplied {@link Metadata} of the request
+   * @return A future that completes when all the chunks are downloaded and written successfully to the OutputStream
+   * @throws org.apache.samza.storage.blobstore.exceptions.DeletedException returned future should complete
+   *         exceptionally with DeletedException on failure with the blob already deleted error.
+   */
+  CompletionStage<Void> get(String id, OutputStream outputStream, Metadata metadata);
+
+  /**
+   * Non-blocking call to mark a blob for deletion in the remote blob store
+   * @param id Blob ID of the blob to delete
+   * @param metadata User supplied {@link Metadata} of the request
+   * @return A future that completes when the blob is successfully deleted from the blob store.
+   * @throws org.apache.samza.storage.blobstore.exceptions.DeletedException returned future should complete
+   *         exceptionally with DeletedException on failure with the blob already deleted error. This exception is
+   *         caught and ignored by the caller of the delete method during initial cleanup and SnapshotIndex read.
+   */
+  CompletionStage<Void> delete(String id, Metadata metadata);
+
+  /**
+   * Non-blocking call to remove the Time-To-Live (TTL) for a blob and make it permanent.
+   * @param blobId Blob ID of blob to remove TTL for.
+   * @param metadata User supplied {@link Metadata} of the request
+   * @return a future that completes when the TTL for the blob is removed.
+   * @throws org.apache.samza.storage.blobstore.exceptions.DeletedException returned future should complete
+   *         exceptionally with DeletedException on failure with the blob already deleted error.
+   */
+  CompletionStage<Void> removeTTL(String blobId, Metadata metadata);
+
+  /**
+   * Cleanly close resources like blob store client
+   */
+  void close();
+}

samza-api/src/main/java/org/apache/samza/storage/blobstore/BlobStoreManagerFactory.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.storage.blobstore;
+
+import java.util.concurrent.ExecutorService;
+import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.config.Config;
+
+@InterfaceStability.Unstable
+public interface BlobStoreManagerFactory {
+  BlobStoreManager getBackupBlobStoreManager(Config config, ExecutorService backupExecutor);
+
+  BlobStoreManager getRestoreBlobStoreManager(Config config, ExecutorService restoreExecutor);
+}

samza-api/src/main/java/org/apache/samza/storage/blobstore/Metadata.java

@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.storage.blobstore;
+
+import java.util.Optional;
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+
+
+/**
+ * Metadata associated with every BlobStore request. This class is used to trace a request and to determine the
+ * bucket/container informationof the blob.
+ */
+public class Metadata {
+  public static final String SNAPSHOT_INDEX_PAYLOAD_PATH = "snapshot-index";
+
+  private final String payloadPath;
+  private final long payloadSize;
+  private final String jobName;
+  private final String jobId;
+  private final String taskName;
+  private final String storeName;
+
+  public Metadata(String payloadPath, Optional<Long> payloadSize,
+      String jobName, String jobId, String taskName, String storeName) {
+    this.payloadPath = payloadPath;
+    // Payload size may not be known in advance for requests like getSnapshotIndex, where only blob ID is known. Set -1.
+    this.payloadSize = payloadSize.orElse(-1L);
+    this.jobName = jobName;
+    this.jobId = jobId;
+    this.taskName = taskName;
+    this.storeName = storeName;
+  }
+
+  public String getPayloadPath() {
+    return payloadPath;
+  }
+
+  public long getPayloadSize() {
+    return payloadSize;
+  }
+
+  public String getJobName() {
+    return jobName;
+  }
+
+  public String getJobId() {
+    return jobId;
+  }
+
+  public String getTaskName() {
+    return taskName;
+  }
+
+  public String getStoreName() {
+    return storeName;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (!(o instanceof Metadata)) {
+      return false;
+    }
+
+    Metadata that = (Metadata) o;
+
+    return new EqualsBuilder().append(getPayloadPath(), that.getPayloadPath())
+        .append(getPayloadSize(), that.getPayloadSize())
+        .append(getJobName(), that.getJobName())
+        .append(getJobId(), that.getJobId())
+        .append(getTaskName(), that.getTaskName())
+        .append(getStoreName(), that.getStoreName())
+        .isEquals();
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder(17, 37).append(getPayloadPath())
+        .append(getPayloadSize())
+        .append(getJobName())
+        .append(getJobId())
+        .append(getTaskName())
+        .append(getStoreName())
+        .toHashCode();
+  }
+
+  @Override
+  public String toString() {
+    return "Metadata{" + "payloadPath='" + payloadPath + '\'' + ", payloadSize='" + payloadSize + '\''
+        + ", jobName='" + jobName + '\'' + ", jobId='" + jobId + '\'' + ", taskName='" + taskName + '\''
+        + ", storeName='" + storeName + '\'' + '}';
+  }
+}

samza-api/src/main/java/org/apache/samza/storage/blobstore/exceptions/DeletedException.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.storage.blobstore.exceptions;
+
+/**
+ * Future should complete with this exception to indicate that the exception occurred due to the request for an
+ * already deleted blob. This exception is caught and ignored by caller of the DELETE request during initial cleanup
+ * and snapshot index read.
+ *
+ */
+public class DeletedException extends RuntimeException {
+
+  private static final long serialVersionUID = 1L;
+
+  public DeletedException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public DeletedException(String message) {
+    super(message);
+  }
+
+  public DeletedException(Throwable cause) {
+    super(cause);
+  }
+
+  public DeletedException() {
+  }
+}