public void register() {
    if (isStarted) {
      log.info("Coordinator stream partition {} has already been registered. Skipping.", coordinatorSystemStreamPartition);
      return;
    }
    log.debug("Attempting to register: {}", coordinatorSystemStreamPartition);
    Set<String> streamNames = new HashSet<String>();
    String streamName = coordinatorSystemStreamPartition.getStream();
    streamNames.add(streamName);
    Map<String, SystemStreamMetadata> systemStreamMetadataMap = systemAdmin.getSystemStreamMetadata(streamNames);
    log.info(String.format("Got metadata %s", systemStreamMetadataMap.toString()));

    if (systemStreamMetadataMap == null) {
      throw new SamzaException("Received a null systemStreamMetadataMap from the systemAdmin. This is illegal.");
    }

    SystemStreamMetadata systemStreamMetadata = systemStreamMetadataMap.get(streamName);

    if (systemStreamMetadata == null) {
      throw new SamzaException("Expected " + streamName + " to be in system stream metadata.");
    }

    SystemStreamPartitionMetadata systemStreamPartitionMetadata = systemStreamMetadata.getSystemStreamPartitionMetadata().get(coordinatorSystemStreamPartition.getPartition());

    if (systemStreamPartitionMetadata == null) {
      throw new SamzaException("Expected metadata for " + coordinatorSystemStreamPartition + " to exist.");
    }

    String startingOffset = systemStreamPartitionMetadata.getOldestOffset();
    log.debug("Registering {} with offset {}", coordinatorSystemStreamPartition, startingOffset);
    systemConsumer.register(coordinatorSystemStreamPartition, startingOffset);
  }
Retrieves the oldest offset in the coordinator stream, and registers the
coordinator stream with the SystemConsumer using the earliest offset.
samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamSystemConsumer.java