public void join(String version, String participantId) {
    LOG.info("Joining the barrier version: {} as participant: {}.", version, participantId);
    String barrierStatePath = keyBuilder.getBarrierStatePath(version);
    LOG.info("Subscribing data changes on the path: {} for barrier version: {}.", barrierStatePath, version);
    zkUtils.subscribeDataChanges(barrierStatePath, new ZkBarrierReachedHandler(barrierStatePath, version, zkUtils));

    // TODO: Handle ZkNodeExistsException - SAMZA-1304
    zkUtils.getZkClient().createPersistent(
        String.format("%s/%s", keyBuilder.getBarrierParticipantsPath(version), participantId));
  }
Joins a shared barrier by registering under the barrier sub-tree in ZK

@param version Version associated with the Barrier
@param participantId Identifier of the participant
samza-core/src/main/java/org/apache/samza/zk/ZkBarrierForVersionUpgrade.java