helix-core/src/main/java/org/apache/helix/manager/zk/ZkStateChangeListener.java

package org.apache.helix.manager.zk;

/*
 * 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.
 */

import java.util.LinkedList;
import java.util.List;

import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.log4j.Logger;
import org.apache.zookeeper.Watcher.Event.KeeperState;

public class ZkStateChangeListener implements IZkStateListener
{
  private volatile boolean _isConnected;
  private volatile boolean _hasSessionExpired;
  private final ZKHelixManager _zkHelixManager;
  
  // Keep track of timestamps that zk State has become Disconnected
  // If in a _timeWindowLengthMs window zk State has become Disconnected 
  // for more than_maxDisconnectThreshold times disconnect the zkHelixManager
  List<Long> _disconnectTimeHistory = new LinkedList<Long>();
  int _timeWindowLengthMs;
  int _maxDisconnectThreshold;

  private static Logger logger = Logger.getLogger(ZkStateChangeListener.class);

  public ZkStateChangeListener(ZKHelixManager zkHelixManager, int timeWindowLengthMs, int maxDisconnectThreshold)
  {
    this._zkHelixManager = zkHelixManager;
    _timeWindowLengthMs = timeWindowLengthMs;
    // _maxDisconnectThreshold min value is 1. 
    // We don't want to disconnect from zk for the first time zkState become Disconnected
    _maxDisconnectThreshold = maxDisconnectThreshold > 0 ? maxDisconnectThreshold : 1;
  }

  @Override
  public void handleNewSession()
  {
    // TODO:bug in zkclient .
    // zkclient does not invoke handleStateChanged when a session expires but
    // directly invokes handleNewSession
    _isConnected = true;
    _hasSessionExpired = false;
    _zkHelixManager.handleNewSession();
  }

  @Override
  public void handleStateChanged(KeeperState keeperState) throws Exception
  {
    switch (keeperState)
    {
    case SyncConnected:
      ZkConnection zkConnection =
          ((ZkConnection) _zkHelixManager._zkClient.getConnection());
      logger.info("KeeperState: " + keeperState + ", zookeeper:" + zkConnection.getZookeeper());
      _isConnected = true;
      break;
    case Disconnected:
      logger.info("KeeperState:" + keeperState + ", disconnectedSessionId: "
          + _zkHelixManager._sessionId + ", instance: "
          + _zkHelixManager.getInstanceName() + ", type: "
          + _zkHelixManager.getInstanceType());

      _isConnected = false;
      // Track the time stamp that the disconnected happens, then check history and see if
      // we should disconnect the _zkHelixManager
      _disconnectTimeHistory.add(System.currentTimeMillis());
      if(isFlapping())
      {
        logger.error("isFlapping() returns true, so disconnect the helix manager. " + _zkHelixManager.getInstanceName() + " "
          + _maxDisconnectThreshold + " disconnects in " + _timeWindowLengthMs + " Ms."); 
        _zkHelixManager.disconnectInternal();
      }
      break;
    case Expired:
      logger.info("KeeperState:" + keeperState + ", expiredSessionId: "
          + _zkHelixManager._sessionId + ", instance: "
          + _zkHelixManager.getInstanceName() + ", type: "
          + _zkHelixManager.getInstanceType());

      _isConnected = false;
      _hasSessionExpired = true;
      break;
    }
  }

  boolean isConnected()
  {
    return _isConnected;
  }

  void disconnect()
  {
    _isConnected = false;
  }

  boolean hasSessionExpired()
  {
    return _hasSessionExpired;
  }
  
  /**
   * If zk state has changed into Disconnected for _maxDisconnectThreshold times during previous _timeWindowLengthMs Ms
   * time window, we think that there are something wrong going on and disconnect the zkHelixManager from zk.
   * */
  boolean isFlapping()
  {
    if(_disconnectTimeHistory.size() == 0)
    {
      return false;
    }
    long mostRecentTimestamp = _disconnectTimeHistory.get(_disconnectTimeHistory.size() - 1);
    // Remove disconnect history timestamp that are older than _timeWindowLengthMs ago
    while((_disconnectTimeHistory.get(0) + _timeWindowLengthMs) < mostRecentTimestamp)
    {
      _disconnectTimeHistory.remove(0);
    }
    return _disconnectTimeHistory.size() > _maxDisconnectThreshold;
  }
}