/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.samza.zk;

import org.apache.samza.config.ZkConfig;
import org.apache.samza.coordinator.CoordinationUtils;
import org.apache.samza.coordinator.Latch;
import org.apache.samza.coordinator.LeaderElector;


public class ZkCoordinationUtils implements CoordinationUtils {
  public final ZkConfig zkConfig;
  public final ZkUtils zkUtils;
  public final String processorIdStr;

  public ZkCoordinationUtils(String processorId, ZkConfig zkConfig, ZkUtils zkUtils) {
    this.zkConfig = zkConfig;
    this.zkUtils = zkUtils;
    this.processorIdStr = processorId;
  }

  @Override
  public void reset() {
    zkUtils.close();
  }

  @Override
  public LeaderElector getLeaderElector() {
    return new ZkLeaderElector(processorIdStr, zkUtils);
  }

  @Override
  public Latch getLatch(int size, String latchId) {
    return new ZkProcessorLatch(size, latchId, processorIdStr, zkUtils);
  }

  // TODO - SAMZA-1128 CoordinationService should directly depend on ZkUtils and DebounceTimer
  public ZkUtils getZkUtils() {
    return zkUtils;
  }
}

@@ -831,16 +831,77 @@
 za.zk;%0A%0A
+import org.I0Itec.zkclient.exception.ZkInterruptedException;%0A
 import o
@@ -933,16 +933,16 @@
 Config;%0A
-
 import o
@@ -1082,16 +1082,73 @@
 lector;%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A
 %0A%0Apublic
@@ -1205,16 +1205,105 @@
 Utils %7B%0A
+  private static final Logger LOG = LoggerFactory.getLogger(ZkCoordinationUtils.class);%0A%0A
   public
@@ -1620,24 +1620,36 @@
 d reset() %7B%0A
+    try %7B%0A  
     zkUtils.
@@ -1657,16 +1657,195 @@
 lose();%0A
+    %7D catch (ZkInterruptedException ex) %7B%0A      // Swallowing due to occurrence in the last stage of lifecycle(Not actionable).%0A      LOG.error(%22Exception in reset: %22, ex);%0A    %7D%0A
   %7D%0A%0A  @

Handle ZKExceptions in zkCoordinationUtils.reset.