@SuppressWarnings("unchecked") private void rebalanceLoadAcrossECWithAffinity(List<JetstreamTopic> topics,Long affinityKey){
  Iterator<JetstreamTopic> itr=topics.iterator();
  while (itr.hasNext()) {
    JetstreamTopic topic=itr.next();
    if (m_eventTopicRegistry.containsKey(topic)) {
      ArrayList<LinkedList<Selection>> weightedeclist=m_eventTopicRegistry.get(topic);
      Iterator<LinkedList<Selection>> weightedeclistitr=weightedeclist.iterator();
      while (weightedeclistitr.hasNext()) {
        LinkedList<Selection> eclist=weightedeclistitr.next();
        Iterator<Selection> eclistitr=eclist.iterator();
        while (eclistitr.hasNext()) {
          EventConsumerInfo info=(EventConsumerInfo)eclistitr.next();
          if (info == null)           continue;
          if (info.containsMultipleAffinityKeyBindings()) {
            if (!info.isAffinityKeyBound(affinityKey))             continue;
            info.unbindAffinityKey(affinityKey);
            Map<Channel,ConsumerChannelContext> contextMap=info.getConsumerChannelContexts();
            Collection<ConsumerChannelContext> consumerContexts=contextMap.values();
            for (            ConsumerChannelContext channelContext : consumerContexts) {
              List<Long> affinityKeyList=(List<Long>)channelContext.getAttribute("affinityKey");
              printInfo("In rebalance - removing key = " + affinityKey.longValue());
              affinityKeyList.remove(affinityKey);
            }
          }
        }
      }
    }
  }
}

@ SuppressWarnings ( "unchecked" ) private void rebalanceLoadAcrossECWithAffinity ( List < JetstreamTopic > topics , Long affinityKey ) { Iterator < JetstreamTopic > itr = topics . iterator ( ) ; while ( itr . hasNext ( ) ) { JetstreamTopic topic = itr . next ( ) ; if ( m_eventTopicRegistry . containsKey ( topic ) ) { ArrayList < LinkedList < Selection > > weightedeclist = m_eventTopicRegistry . get ( topic ) ; Iterator < LinkedList < Selection > > weightedeclistitr = weightedeclist . iterator ( ) ; while ( weightedeclistitr . hasNext ( ) ) { LinkedList < Selection > eclist = weightedeclistitr . next ( ) ; Iterator < Selection > eclistitr = eclist . iterator ( ) ; while ( eclistitr . hasNext ( ) ) { EventConsumerInfo info = ( EventConsumerInfo ) eclistitr . next ( ) ; if ( info == null ) continue ; if ( info . containsMultipleAffinityKeyBindings ( ) ) { if ( ! info . isAffinityKeyBound ( affinityKey ) ) continue ; info . unbindAffinityKey ( affinityKey ) ; Map < Channel , ConsumerChannelContext > contextMap = info . getConsumerChannelContexts ( ) ; Collection < ConsumerChannelContext > consumerContexts = contextMap . values ( ) ; for ( ConsumerChannelContext channelContext : consumerContexts ) { List < Long > affinityKeyList = ( List < Long > ) channelContext . getAttribute ( "affinityKey" ) ; printInfo ( "In rebalance - removing key = " + affinityKey . longValue ( ) ) ; affinityKeyList . remove ( affinityKey ) ; } } } } } } }
rebalanceloadacrossecwithaffinity - this method is called when a dead consumer comes back alive and is discovered . the traffic for this consumer if directed elesewhere is now moved back to this rediscoved consumer .
