Consolidating offset read and write for store-offsets and side-inputs, maintaining backward compatbility

This consolidates the two different kinds of offsets used by stores and side-inputs into one.
The code can still read and make sense of store-offset files written in the old format.
The new format stores a map of ssp to offset, rather than a singular string.

Updated tests.
And added tests to ensure the reading of the old-format offset still works.

Author: Ray Matharu <[email protected]>

Reviewers: mynameborat

Closes apache#915 from rmatharu/test-offset and squashes the following commits:

c20e94c4 [Ray Matharu] minor
403f72b1 [Ray Matharu] simplifying code
b5868066 [Ray Matharu] minor
ca648223 [Ray Matharu] gradle.props
b50e5fe [Ray Matharu] Applying comments
75905b0c [Ray Matharu] minor
a8d1dcdc [Ray Matharu] Addressing review comments
1f7ecdcc [Ray Matharu] minor
666157ad [Ray Matharu] test fix
d621a51c [Ray Matharu] Adding test for read of old format
b9e58997 [Ray Matharu] build fix
9cb822ca [Ray Matharu] bug fix
5484e430 [Ray Matharu] Consolidating writing of offsets
55f35b7b [Ray Matharu] minor
88963a8c [Ray Matharu] minor
f14ee03b [Ray Matharu] Consolidating read of offset file
12ca96bb [Ray Matharu] Merge branch 'master' of https://github.com/apache/samza
ee7daac8 [Ray Matharu] Merge branch 'master' of https://github.com/apache/samza
08006871 [Ray Matharu] Merge branch 'master' of https://github.com/apache/samza
916f66ae [Ray Matharu] Merge branch 'master' of https://github.com/apache/samza
2c09b081 [Ray Matharu] Rocksdb bug fix

Author: rmatharu

samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java

@@ -21,16 +21,34 @@
 
 import com.google.common.collect.ImmutableMap;
 import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import java.util.stream.Collectors;
 import org.apache.samza.container.TaskName;
+import org.apache.samza.serializers.model.SamzaObjectMapper;
 import org.apache.samza.system.SystemAdmin;
 import org.apache.samza.system.SystemStreamPartition;
 import org.apache.samza.util.FileUtil;
+import org.codehaus.jackson.JsonParseException;
+import org.codehaus.jackson.map.JsonMappingException;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.ObjectWriter;
+import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 
 public class StorageManagerUtil {
   private static final Logger LOG = LoggerFactory.getLogger(StorageManagerUtil.class);
+  public static final String OFFSET_FILE_NAME = "OFFSET";
+  private static final ObjectMapper OBJECT_MAPPER = SamzaObjectMapper.getObjectMapper();
+  private static final TypeReference<Map<SystemStreamPartition, String>> OFFSETS_TYPE_REFERENCE =
+            new TypeReference<Map<SystemStreamPartition, String>>() { };
+  private static final ObjectWriter OBJECT_WRITER = OBJECT_MAPPER.writerWithType(OFFSETS_TYPE_REFERENCE);
+
 
   /**
    * Fetch the starting offset for the input {@link SystemStreamPartition}
@@ -70,17 +88,15 @@ public static String getStartingOffset(
    * the {@code storeDeleteRetentionInMs}, then the store is considered stale.
    *
    * @param storeDir the base directory of the store
-   * @param offsetFileName the offset file name
    * @param storeDeleteRetentionInMs store delete retention in millis
    * @param currentTimeMs current time in ms
    * @return true if the store is stale, false otherwise
    */
-  public static boolean isStaleStore(
-      File storeDir, String offsetFileName, long storeDeleteRetentionInMs, long currentTimeMs) {
+  public static boolean isStaleStore(File storeDir, long storeDeleteRetentionInMs, long currentTimeMs) {
     boolean isStaleStore = false;
     String storePath = storeDir.toPath().toString();
     if (storeDir.exists()) {
-      File offsetFileRef = new File(storeDir, offsetFileName);
+      File offsetFileRef = new File(storeDir, OFFSET_FILE_NAME);
       long offsetFileLastModifiedTime = offsetFileRef.lastModified();
       if ((currentTimeMs - offsetFileLastModifiedTime) >= storeDeleteRetentionInMs) {
         LOG.info(
@@ -98,14 +114,14 @@ public static boolean isStaleStore(
    * An offset file associated with logged store {@code storeDir} is valid if it exists and is not empty.
    *
    * @param storeDir the base directory of the store
-   * @param offsetFileName name of the offset file
+   * @param storeSSPs storeSSPs (if any) associated with the store
    * @return true if the offset file is valid. false otherwise.
    */
-  public static boolean isOffsetFileValid(File storeDir, String offsetFileName) {
+  public static boolean isOffsetFileValid(File storeDir, Set<SystemStreamPartition> storeSSPs) {
     boolean hasValidOffsetFile = false;
     if (storeDir.exists()) {
-      String offsetContents = readOffsetFile(storeDir, offsetFileName);
-      if (offsetContents != null && !offsetContents.isEmpty()) {
+      Map<SystemStreamPartition, String> offsetContents = readOffsetFile(storeDir, storeSSPs);
+      if (offsetContents != null && !offsetContents.isEmpty() && offsetContents.keySet().equals(storeSSPs)) {
         hasValidOffsetFile = true;
       } else {
         LOG.info("Offset file is not valid for store: {}.", storeDir.toPath());
@@ -115,6 +131,34 @@ public static boolean isOffsetFileValid(File storeDir, String offsetFileName) {
     return hasValidOffsetFile;
   }
 
+  /**
+   * Write the given SSP-Offset map into the offsets file.
+   * @param storeBaseDir the base directory of the store
+   * @param storeName the store name to use
+   * @param taskName the task name which is referencing the store
+   * @param offsets The SSP-offset to write
+   * @throws IOException because of deserializing to json
+   */
+  public static void writeOffsetFile(File storeBaseDir, String storeName, TaskName taskName,
+      Map<SystemStreamPartition, String> offsets) throws IOException {
+    File offsetFile = new File(getStorePartitionDir(storeBaseDir, storeName, taskName), OFFSET_FILE_NAME);
+    String fileContents = OBJECT_WRITER.writeValueAsString(offsets);
+    FileUtil.writeWithChecksum(offsetFile, fileContents);
+  }
+
+  /**
+   *  Delete the offset file for this task and store, if one exists.
+   * @param storeBaseDir the base directory of the store
+   * @param storeName the store name to use
+   * @param taskName the task name which is referencing the store
+   */
+  public static void deleteOffsetFile(File storeBaseDir, String storeName, TaskName taskName) {
+    File offsetFile = new File(getStorePartitionDir(storeBaseDir, storeName, taskName), OFFSET_FILE_NAME);
+    if (offsetFile.exists()) {
+      FileUtil.rm(offsetFile);
+    }
+  }
+
   /**
    * Check if a store's disk footprint exists.
    *
@@ -129,26 +173,32 @@ public static boolean storeExists(File storeDir) {
    * Read and return the contents of the offset file.
    *
    * @param storagePartitionDir the base directory of the store
-   * @param offsetFileName name of the offset file
+   * @param storeSSPs SSPs associated with the store (if any)
    * @return the content of the offset file if it exists for the store, null otherwise.
    */
-  public static String readOffsetFile(File storagePartitionDir, String offsetFileName) {
-    String offset = null;
-    File offsetFileRef = new File(storagePartitionDir, offsetFileName);
+  public static Map<SystemStreamPartition, String> readOffsetFile(File storagePartitionDir, Set<SystemStreamPartition> storeSSPs) {
+    Map<SystemStreamPartition, String> offsets = new HashMap<>();
+    String fileContents = null;
+    File offsetFileRef = new File(storagePartitionDir, OFFSET_FILE_NAME);
     String storePath = storagePartitionDir.getPath();
 
     if (offsetFileRef.exists()) {
       LOG.info("Found offset file in storage partition directory: {}", storePath);
       try {
-        offset = FileUtil.readWithChecksum(offsetFileRef);
+        fileContents = FileUtil.readWithChecksum(offsetFileRef);
+        offsets = OBJECT_MAPPER.readValue(fileContents, OFFSETS_TYPE_REFERENCE);
+      } catch (JsonParseException | JsonMappingException e) {
+        LOG.info("Exception in json-parsing offset file {} {}, reading as string offset-value", storagePartitionDir.toPath(), OFFSET_FILE_NAME);
+        final String finalFileContents = fileContents;
+        offsets = (storeSSPs.size() == 1) ? storeSSPs.stream().collect(Collectors.toMap(ssp -> ssp, offset -> finalFileContents)) : offsets;
       } catch (Exception e) {
         LOG.warn("Failed to read offset file in storage partition directory: {}", storePath, e);
       }
     } else {
       LOG.info("No offset file found in storage partition directory: {}", storePath);
     }
 
-    return offset;
+    return offsets;
   }
 
   /**

samza-core/src/main/java/org/apache/samza/storage/TaskSideInputStorageManager.java

@@ -39,7 +39,6 @@
 import org.apache.samza.SamzaException;
 import org.apache.samza.config.Config;
 import org.apache.samza.container.TaskName;
-import org.apache.samza.serializers.model.SamzaObjectMapper;
 import org.apache.samza.storage.kv.Entry;
 import org.apache.samza.storage.kv.KeyValueStore;
 import org.apache.samza.system.IncomingMessageEnvelope;
@@ -51,9 +50,6 @@
 import org.apache.samza.util.Clock;
 import org.apache.samza.util.FileUtil;
 
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.map.ObjectWriter;
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.collection.JavaConverters;
@@ -65,17 +61,12 @@
  */
 public class TaskSideInputStorageManager {
   private static final Logger LOG = LoggerFactory.getLogger(TaskSideInputStorageManager.class);
-  private static final String OFFSET_FILE = "SIDE-INPUT-OFFSETS";
   private static final long STORE_DELETE_RETENTION_MS = TimeUnit.DAYS.toMillis(1); // same as changelog delete retention
-  private static final ObjectMapper OBJECT_MAPPER = SamzaObjectMapper.getObjectMapper();
-  private static final TypeReference<HashMap<SystemStreamPartition, String>> OFFSETS_TYPE_REFERENCE =
-      new TypeReference<HashMap<SystemStreamPartition, String>>() { };
-  private static final ObjectWriter OBJECT_WRITER = OBJECT_MAPPER.writerWithType(OFFSETS_TYPE_REFERENCE);
 
   private final Clock clock;
   private final Map<String, SideInputsProcessor> storeToProcessor;
   private final Map<String, StorageEngine> stores;
-  private final String storeBaseDir;
+  private final File storeBaseDir;
   private final Map<String, Set<SystemStreamPartition>> storeToSSps;
   private final Map<SystemStreamPartition, Set<String>> sspsToStores;
   private final StreamMetadataCache streamMetadataCache;
@@ -97,7 +88,7 @@ public TaskSideInputStorageManager(
       Clock clock) {
     this.clock = clock;
     this.stores = sideInputStores;
-    this.storeBaseDir = storeBaseDir;
+    this.storeBaseDir = new File(storeBaseDir);
     this.storeToSSps = storesToSSPs;
     this.streamMetadataCache = streamMetadataCache;
     this.systemAdmins = systemAdmins;
@@ -258,9 +249,7 @@ void writeOffsetFiles() {
               .collect(Collectors.toMap(Function.identity(), lastProcessedOffsets::get));
 
             try {
-              String fileContents = OBJECT_WRITER.writeValueAsString(offsets);
-              File offsetFile = new File(getStoreLocation(storeName), OFFSET_FILE);
-              FileUtil.writeWithChecksum(offsetFile, fileContents);
+              StorageManagerUtil.writeOffsetFile(storeBaseDir, storeName, taskName, offsets);
             } catch (Exception e) {
               throw new SamzaException("Failed to write offset file for side input store: " + storeName, e);
             }
@@ -284,8 +273,8 @@ Map<SystemStreamPartition, String> getFileOffsets() {
         File storeLocation = getStoreLocation(storeName);
         if (isValidSideInputStore(storeName, storeLocation)) {
           try {
-            String fileContents = StorageManagerUtil.readOffsetFile(storeLocation, OFFSET_FILE);
-            Map<SystemStreamPartition, String> offsets = OBJECT_MAPPER.readValue(fileContents, OFFSETS_TYPE_REFERENCE);
+
+            Map<SystemStreamPartition, String> offsets = StorageManagerUtil.readOffsetFile(storeLocation, storeToSSps.get(storeName));
             fileOffsets.putAll(offsets);
           } catch (Exception e) {
             LOG.warn("Failed to load the offset file for side input store:" + storeName, e);
@@ -298,7 +287,7 @@ Map<SystemStreamPartition, String> getFileOffsets() {
 
   @VisibleForTesting
   File getStoreLocation(String storeName) {
-    return new File(storeBaseDir, (storeName + File.separator + taskName.toString()).replace(' ', '_'));
+    return StorageManagerUtil.getStorePartitionDir(storeBaseDir, storeName, taskName);
   }
 
   /**
@@ -368,8 +357,8 @@ Map<SystemStreamPartition, String> getOldestOffsets() {
 
   private boolean isValidSideInputStore(String storeName, File storeLocation) {
     return isPersistedStore(storeName)
-        && !StorageManagerUtil.isStaleStore(storeLocation, OFFSET_FILE, STORE_DELETE_RETENTION_MS, clock.currentTimeMillis())
-        && StorageManagerUtil.isOffsetFileValid(storeLocation, OFFSET_FILE);
+        && !StorageManagerUtil.isStaleStore(storeLocation, STORE_DELETE_RETENTION_MS, clock.currentTimeMillis())
+        && StorageManagerUtil.isOffsetFileValid(storeLocation, storeToSSps.get(storeName));
   }
 
   private boolean isPersistedStore(String storeName) {

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

@@ -23,6 +23,7 @@
 import java.io.File;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -466,8 +467,6 @@ public Void call() {
    * with the respective consumer, restoring stores, and stopping stores.
    */
   private class TaskRestoreManager {
-
-    private final static String OFFSET_FILE_NAME = "OFFSET";
     private final Map<String, StorageEngine> taskStores; // Map of all StorageEngines for this task indexed by store name
     private final Set<String> taskStoresToRestore;
     // Set of store names which need to be restored by consuming using system-consumers (see registerStartingOffsets)
@@ -532,14 +531,14 @@ private void cleanBaseDirsAndReadOffsetFiles() {
             LOG.info("Deleting logged storage partition directory " + loggedStorePartitionDir.toPath().toString());
             FileUtil.rm(loggedStorePartitionDir);
           } else {
-            String offset = StorageManagerUtil.readOffsetFile(loggedStorePartitionDir, OFFSET_FILE_NAME);
-            LOG.info("Read offset " + offset + " for the store " + storeName + " from logged storage partition directory "
-                + loggedStorePartitionDir);
-
-            if (offset != null) {
-              fileOffsets.put(
-                  new SystemStreamPartition(changelogSystemStreams.get(storeName), taskModel.getChangelogPartition()),
-                  offset);
+
+            SystemStreamPartition changelogSSP = new SystemStreamPartition(changelogSystemStreams.get(storeName), taskModel.getChangelogPartition());
+            Map<SystemStreamPartition, String> offset =
+                StorageManagerUtil.readOffsetFile(loggedStorePartitionDir, Collections.singleton(changelogSSP));
+            LOG.info("Read offset {} for the store {} from logged storage partition directory {}", offset, storeName, loggedStorePartitionDir);
+
+            if (offset.containsKey(changelogSSP)) {
+              fileOffsets.put(changelogSSP, offset.get(changelogSSP));
             }
           }
         });
@@ -562,9 +561,13 @@ private boolean isLoggedStoreValid(String storeName, File loggedStoreDir) {
             (long) new StorageConfig(config).getChangeLogDeleteRetentionsInMs().get(storeName).get();
       }
 
-      return this.taskStores.get(storeName).getStoreProperties().isPersistedToDisk()
-          && StorageManagerUtil.isOffsetFileValid(loggedStoreDir, OFFSET_FILE_NAME) && !StorageManagerUtil.isStaleStore(
-          loggedStoreDir, OFFSET_FILE_NAME, changeLogDeleteRetentionInMs, clock.currentTimeMillis());
+      if (changelogSystemStreams.containsKey(storeName)) {
+        SystemStreamPartition changelogSSP = new SystemStreamPartition(changelogSystemStreams.get(storeName), taskModel.getChangelogPartition());
+        return this.taskStores.get(storeName).getStoreProperties().isPersistedToDisk() && StorageManagerUtil.isOffsetFileValid(loggedStoreDir, Collections.singleton(changelogSSP))
+            && !StorageManagerUtil.isStaleStore(loggedStoreDir, changeLogDeleteRetentionInMs, clock.currentTimeMillis());
+      }
+
+      return false;
     }
 
     /**

samza-core/src/main/scala/org/apache/samza/storage/TaskStorageManager.scala

@@ -45,8 +45,6 @@ class TaskStorageManager(
     case (storeName, storageEngine) => storageEngine.getStoreProperties.isPersistedToDisk
   }
 
-  val offsetFileName = "OFFSET"
-
   def getStore(storeName: String): Option[StorageEngine] =  JavaOptionals.toRichOptional(containerStorageManager.getStore(taskName, storeName)).toOption
 
   def init {
@@ -90,17 +88,13 @@ class TaskStorageManager(
         val newestOffset = if (sspMetadata == null) null else sspMetadata.getNewestOffset
         debug("Got offset %s for store %s" format(newestOffset, storeName))
 
-        val loggedStorePartitionDir = StorageManagerUtil.getStorePartitionDir(loggedStoreBaseDir, storeName, taskName)
-        val offsetFile = new File(loggedStorePartitionDir, offsetFileName)
         if (newestOffset != null) {
           debug("Storing offset for store in OFFSET file ")
-          FileUtil.writeWithChecksum(offsetFile, newestOffset)
+          StorageManagerUtil.writeOffsetFile(loggedStoreBaseDir, storeName, taskName, Map(ssp -> newestOffset).asJava)
           debug("Successfully stored offset %s for store %s in OFFSET file " format(newestOffset, storeName))
         } else {
           //if newestOffset is null, then it means the store is (or has become) empty. No need to persist the offset file
-          if (offsetFile.exists()) {
-            FileUtil.rm(offsetFile)
-          }
+          StorageManagerUtil.deleteOffsetFile(loggedStoreBaseDir, storeName, taskName);
           debug("Not storing OFFSET file for taskName %s. Store %s backed by changelog topic: %s, partition: %s is empty. " format (taskName, storeName, systemStream.getStream, partition.getPartitionId))
         }
       } catch {