@Override public synchronized boolean takePartition(String topic,int partition){
  Set<Integer> partitionSet=m_myTopicPartitions.get(topic);
  if (partitionSet == null)   return false;
  String ownerPath=ownerPath(topic,partition);
  try {
    if (m_zk.exists(ownerPath)) {
      return false;
    }
    if (m_zk.create(ownerPath,true)) {
      m_zk.writeString(ownerPath,m_consumerId);
      partitionSet.add(partition);
      return true;
    }
 else     return false;
  }
 catch (  RuntimeException e) {
    throw e;
  }
catch (  Exception e) {
    LOGGER.error("Fail to take partition " + ownerPath,e);
    return false;
  }
}

@ Override public synchronized boolean takePartition ( String topic , int partition ) { Set < Integer > partitionSet = m_myTopicPartitions . get ( topic ) ; if ( partitionSet == null ) return false ; String ownerPath = ownerPath ( topic , partition ) ; try { if ( m_zk . exists ( ownerPath ) ) { return false ; } if ( m_zk . create ( ownerPath , true ) ) { m_zk . writeString ( ownerPath , m_consumerId ) ; partitionSet . add ( partition ) ; return true ; } else return false ; } catch ( RuntimeException e ) { throw e ; } catch ( Exception e ) { LOGGER . error ( "Fail to take partition " + ownerPath , e ) ; return false ; } }
take the partition in zookeeper , if taken ( not idle ) , return false
