question: @ 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 ; } }
target:  take the partition in zookeeper, if taken(not idle), return false