/**
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.hadoop.metrics2.sink.timeline.availability;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.CuratorZookeeperClient;
import org.apache.curator.RetryLoop;
import org.apache.curator.RetryPolicy;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.ZooKeeper;

import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;

/**
 * Find a live Collector instance from Zookeeper
 * This class allows connect to ZK on-demand and
 * does not add a watcher on the znode.
 */
public class MetricCollectorHAHelper {
  private final String zookeeperQuorum;
  private final int tryTime;
  private final int sleepMsBetweenRetries;

  private static final int CONNECTION_TIMEOUT = 2000;
  private static final int SESSION_TIMEOUT = 10000;
  private static final String ZK_PATH = "/ambari-metrics-cluster/LIVEINSTANCES";
  private static final String INSTANCE_NAME_DELIMITER = "_";



  private static final Log LOG = LogFactory.getLog(MetricCollectorHAHelper.class);

  public MetricCollectorHAHelper(String zookeeperQuorum, int tryTime, int sleepMsBetweenRetries) {
    this.zookeeperQuorum = zookeeperQuorum;
    this.tryTime = tryTime;
    this.sleepMsBetweenRetries = sleepMsBetweenRetries;
  }

  /**
   * Connect to Zookeeper to find live instances of metrics collector
   * @return {#link Collection} hostnames
   */
  public Collection<String> findLiveCollectorHostsFromZNode() {
    Set<String> collectors = new HashSet<>();

    RetryPolicy retryPolicy = new RetryUntilElapsed(tryTime, sleepMsBetweenRetries);
    final CuratorZookeeperClient client = new CuratorZookeeperClient(zookeeperQuorum,
      SESSION_TIMEOUT, CONNECTION_TIMEOUT, null, retryPolicy);

    List<String> liveInstances = null;

    try {
      client.start();
      liveInstances = RetryLoop.callWithRetry(client, new Callable<List<String>>() {
        @Override
        public List<String> call() throws Exception {
          ZooKeeper zookeeper = client.getZooKeeper();
          return zookeeper.getChildren(ZK_PATH, false);
        }
      });
    } catch (Exception e) {
      LOG.warn("Unable to connect to zookeeper.", e);
      LOG.debug(e);
    }

    // [ambari-sid-3.c.pramod-thangali.internal_12001]
    if (liveInstances != null && !liveInstances.isEmpty()) {
      for (String instanceStr : liveInstances) {
        collectors.add(instanceStr.substring(0, instanceStr.indexOf(INSTANCE_NAME_DELIMITER)));
      }
    }

    return collectors;
  }
}

@@ -1081,16 +1081,62 @@
 Policy;%0A
+import org.apache.curator.retry.RetryOneTime;%0A
 import o
@@ -1218,16 +1218,51 @@
 eeper;%0A%0A
+import java.net.HttpURLConnection;%0A
 import j
@@ -1312,31 +1312,8 @@
 et;%0A
-import java.util.List;%0A
 impo
@@ -1770,10 +1770,9 @@
 T = 
-10
+5
 000;
@@ -2707,28 +2707,22 @@
 );%0A%0A    
-List%3C
 String
-%3E
  liveIns
@@ -2751,30 +2751,8 @@
 y %7B%0A
-      client.start();%0A
     
@@ -2818,21 +2818,15 @@
 ble%3C
-List%3C
 String%3E
-%3E
 () %7B
@@ -2863,20 +2863,14 @@
 lic 
-List%3C
 String
-%3E
  cal
@@ -2957,22 +2957,29 @@
         
-return
+byte%5B%5D data =
  zookeep
@@ -2988,16 +2988,12 @@
 .get
-Children
+Data
 (ZK_
@@ -3002,14 +3002,76 @@
 TH, 
-false)
+null, null);%0A          return data != null ? new String(data) : null
 ;%0A  
@@ -3358,16 +3358,27 @@
 nstances
+.split(%22,%22)
 ) %7B%0A    

Revert "AMBARI-18012 : Metrics Sink unable to connect to zookeeper to locate collector host. (avijayan)"