public void bootstrap() {
    synchronized (bootstrapLock) {
      // Make a copy so readers aren't affected while we modify the set.
      final LinkedHashSet<CoordinatorStreamMessage> bootstrappedMessages = new LinkedHashSet<>(bootstrappedStreamSet);

      log.info("Bootstrapping configuration from coordinator stream.");
      SystemStreamPartitionIterator iterator =
          new SystemStreamPartitionIterator(systemConsumer, coordinatorSystemStreamPartition);

      try {
        while (iterator.hasNext()) {
          IncomingMessageEnvelope envelope = iterator.next();
          Object[] keyArray = keySerde.fromBytes((byte[]) envelope.getKey()).toArray();
          Map<String, Object> valueMap = null;
          if (envelope.getMessage() != null) {
            valueMap = messageSerde.fromBytes((byte[]) envelope.getMessage());
          }
          CoordinatorStreamMessage coordinatorStreamMessage = new CoordinatorStreamMessage(keyArray, valueMap);
          log.debug("Received coordinator stream message: {}", coordinatorStreamMessage);
          // Remove any existing entry. Set.add() does not add if the element already exists.
          if (bootstrappedMessages.remove(coordinatorStreamMessage)) {
            log.debug("Removed duplicate message: {}", coordinatorStreamMessage);
          }
          bootstrappedMessages.add(coordinatorStreamMessage);
          if (SetConfig.TYPE.equals(coordinatorStreamMessage.getType())) {
            String configKey = coordinatorStreamMessage.getKey();
            if (coordinatorStreamMessage.isDelete()) {
              configMap.remove(configKey);
            } else {
              String configValue = new SetConfig(coordinatorStreamMessage).getConfigValue();
              configMap.put(configKey, configValue);
            }
          }
        }

        bootstrappedStreamSet = Collections.unmodifiableSet(bootstrappedMessages);
        log.debug("Bootstrapped configuration: {}", configMap);
        isBootstrapped = true;
      } catch (Exception e) {
        throw new SamzaException(e);
      }
    }
  }
Read all messages from the earliest offset, all the way to the latest.
Currently, this method only pays attention to config messages.
samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamSystemConsumer.java