/**
 * Copyright (C) 2014-2016 LinkedIn Corp. (pinot-core@linkedin.com)
 *
 * Licensed 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 com.linkedin.pinot.broker.broker.helix;

import com.linkedin.pinot.common.response.ServerInstance;
import com.linkedin.pinot.common.utils.CommonConstants;
import com.linkedin.pinot.transport.netty.PooledNettyClientResourceManager;
import com.linkedin.pinot.transport.pool.KeyedPool;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.helix.LiveInstanceChangeListener;
import org.apache.helix.NotificationContext;
import org.apache.helix.model.LiveInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class LiveInstancesChangeListenerImpl implements LiveInstanceChangeListener {

  private static final Logger LOGGER = LoggerFactory.getLogger(LiveInstancesChangeListenerImpl.class);

  private static final boolean DO_NOT_RECREATE = false;

  private long timeout;
  private final Map<String, String> liveInstanceToSessionIdMap;
  private KeyedPool<PooledNettyClientResourceManager.PooledClientConnection> connectionPool;

  public LiveInstancesChangeListenerImpl(String clusterName) {
    this.liveInstanceToSessionIdMap = new HashMap<String, String>();
  }

  public void init(final KeyedPool<PooledNettyClientResourceManager.PooledClientConnection> connectionPool, final long timeout) {
    this.connectionPool = connectionPool;
    this.timeout = timeout;
  }

  @Override
  public void onLiveInstanceChange(List<LiveInstance> liveInstances, NotificationContext changeContext) {
    if (connectionPool == null) {
      LOGGER.info("init hasn't been called yet on the live instances listener...");
      return;
    }

    LOGGER.info("Connection pool found, moving on...");

    for (LiveInstance instance : liveInstances) {

      String instanceId = instance.getInstanceName();
      String sessionId = instance.getSessionId();

      if (!instanceId.startsWith(CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE)) {
        LOGGER.info("Skipping non-server instance: {}", instanceId);
        continue;
      }

      String namePortStr = instanceId.split(CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE)[1];
      String hostName = namePortStr.split("_")[0];
      int port;
      try {
        port = Integer.parseInt(namePortStr.split("_")[1]);
      } catch (Exception e) {
        LOGGER.warn("Port for server instance " + instanceId + " does not appear to be numeric", e);
        port = CommonConstants.Helix.DEFAULT_SERVER_NETTY_PORT;
      }

      if (liveInstanceToSessionIdMap.containsKey(instanceId)) {
        // sessionId has changed
        LOGGER.info("found instance Id : {} with new session Id : {} old session Id {}", instanceId, sessionId,
            liveInstanceToSessionIdMap.get(instanceId));
        if (!sessionId.equals(liveInstanceToSessionIdMap.get(instanceId))) {
          try {
            ServerInstance ins = ServerInstance.forHostPort(hostName, port);
            connectionPool.validatePool(ins, DO_NOT_RECREATE);
            liveInstanceToSessionIdMap.put(instanceId, sessionId);
          } catch (Exception e) {
            LOGGER.error("Error trying to validate & destroy dead connections for {}", instanceId, e);
          }
        }
      } else {
        LOGGER.info("found instance Id : {} with new session Id : {}", instanceId, sessionId);
        // we don't have this instanceId
        // lets first check if the connection is valid or not
        try {
          ServerInstance ins = ServerInstance.forHostPort(hostName, port);
          connectionPool.validatePool(ins, DO_NOT_RECREATE);
          liveInstanceToSessionIdMap.put(instanceId, sessionId);
        } catch (Exception e) {
          LOGGER.error("Error trying to destroy dead connections for {}", instanceId, e);
        }
      }
    }
  }
}

@@ -2129,20 +2129,20 @@
  LOGGER.
-info
+warn
 (%22init h
@@ -2143,18 +2143,19 @@
 init has
-n'
+ no
 t been c
@@ -2164,12 +2164,8 @@
 led 
-yet 
 on t
@@ -2194,91 +2194,63 @@
 ener
-...%22);%0A      return;%0A    %7D%0A%0A    LOGGER.info(%22Connection pool found, moving on...%22);
+, ignoring live instance change.%22);%0A      return;%0A    %7D
 %0A%0A  
@@ -2503,20 +2503,21 @@
  LOGGER.
-info
+debug
 (%22Skippi
@@ -2538,17 +2538,16 @@
 instance
-:
  %7B%7D%22, in
@@ -2842,32 +2842,96 @@
 (Exception e) %7B%0A
+        port = CommonConstants.Helix.DEFAULT_SERVER_NETTY_PORT;%0A
         LOGGER.w
@@ -2964,26 +2964,10 @@
 nce 
-%22 + instanceId + %22
+%7B%7D
  doe
@@ -2996,77 +2996,50 @@
 eric
-%22
 , 
-e);%0A        port = CommonConstants.Helix.DEFAULT_SERVER_NETTY_PORT
+defaulting to %7B%7D.%22, instanceId, port, e)
 ;%0A  
@@ -3154,52 +3154,141 @@
     
-LOGGER.info(%22found instance Id : %7B%7D with new
+if (!sessionId.equals(liveInstanceToSessionIdMap.get(instanceId))) %7B%0A          try %7B%0A            LOGGER.info(%22Instance %7B%7D has changed
  ses
@@ -3296,146 +3296,108 @@
 ion 
-Id : %7B%7D old session Id %7B%7D%22, instanceId, sessionId,%0A            liveInstanceToSessionIdMap.get(instanceId));%0A        if (!sessionId.equals(
+id %7B%7D -%3E %7B%7D, validating connection pool for this instance.%22, instanceId, sessionId,%0A                
 live
@@ -3435,35 +3435,17 @@
 anceId))
-) %7B%0A          try %7B
+;
 %0A       
@@ -3843,22 +3843,26 @@
 fo(%22
-f
+F
 ound 
+new 
 instance
  Id 
@@ -3857,21 +3857,16 @@
 instance
- Id :
  %7B%7D with
@@ -3870,20 +3870,16 @@
 ith 
-new 
 session 
 Id :
@@ -3878,15 +3878,40 @@
 ion 
-I
+i
 d 
-: 
 %7B%7D
+, adding to session id map.
 %22, i

Clean up logging for the live instances listener (#1495)