@Override
  public Checkpoint readLastCheckpoint(TaskName taskName) {
    if (!taskNames.contains(taskName)) {
      throw new SamzaException("reading checkpoint of unregistered/unwritten task");
    }

    // Create the query for taskName
    String partitionQueryKey = taskName.toString();
    String partitionFilter = TableQuery.generateFilterCondition(
            PARTITION_KEY,
            TableQuery.QueryComparisons.EQUAL,
            partitionQueryKey);
    TableQuery<TaskCheckpointEntity> query = TableQuery.from(TaskCheckpointEntity.class).where(partitionFilter);

    ImmutableMap.Builder<SystemStreamPartition, String> builder = ImmutableMap.builder();
    try {
      for (TaskCheckpointEntity taskCheckpointEntity : cloudTable.execute(query)) {
        // Recreate the SSP offset
        String serializedSSP = taskCheckpointEntity.getRowKey();
        builder.put(deserializeSystemStreamPartition(serializedSSP), taskCheckpointEntity.getOffset());
      }

    } catch (NoSuchElementException e) {
      LOG.warn("No checkpoints found found for registered taskName={}", taskName);
      // Return null if not entity elements are not found
      return null;
    }
    LOG.debug("Received checkpoint state for taskName=%s", taskName);
    return new CheckpointV1(builder.build());
  }
Azure checkpoint manager is used to store checkpoints in a Azure Table.
All the task checkpoints are added to the a single table named "SamzaTaskCheckpoints".
The table entities take the following form:

+-----------------+---------------------+-------------------+
|                 |     Serialized      |                   |
|   TaskName      |     JSON SSP        |     Offset        |
|                 |                     |                   |
+-----------------+---------------------+-------------------+

 Each entity have a partitionKey set as the TaskName and the rowKey set as the SSP.
samza-azure/src/main/java/org/apache/samza/checkpoint/azure/AzureCheckpointManager.java