public void validateZkVersion() {

    // Version of the protocol is written into root znode. If root does not exist yet we need to create one.
    String rootPath = keyBuilder.getRootPath();
    if (!zkClient.exists(rootPath)) {
      try {
        // attempt to create the root with the correct version
        zkClient.createPersistent(rootPath, ZK_PROTOCOL_VERSION);
        LOG.info("Created zk root node: " + rootPath + " with zk version " + ZK_PROTOCOL_VERSION);
        return;
      } catch (ZkNodeExistsException e) {
        // ignoring
        LOG.warn("root path " + rootPath + " already exists.");
      }
    }
    // if exists, verify the version
    Stat stat = new Stat();
    String version = zkClient.readData(rootPath, stat);
    if (version == null) {
      // for backward compatibility, if no value - assume 1.0
      try {
        zkClient.writeData(rootPath, "1.0", stat.getVersion());
      } catch (ZkBadVersionException e) {
        // if the write failed with ZkBadVersionException it means someone else already wrote a version, so we can ignore it.
      }
      // re-read the updated version
      version = zkClient.readData(rootPath);
    }
    LOG.info("Current version for zk root node: " + rootPath + " is " + version + ", expected version is " + ZK_PROTOCOL_VERSION);
    if (!version.equals(ZK_PROTOCOL_VERSION)) {
      throw new SamzaException("ZK Protocol mismatch. Expected " + ZK_PROTOCOL_VERSION + "; found " + version);
    }
  }
Publish to most recent job model version that is agreed by the quorum.
@param version active job model version
samza-core/src/main/java/org/apache/samza/zk/ZkUtils.java