private boolean isPartitionRequired(int opid,List<KafkaConsumer.KafkaMeterStats> kstats){
  long t=System.currentTimeMillis();
  if (kstats.size() > 0) {
    logger.debug("Checking offset updates for offset manager");
    updateOffsets(kstats);
  }
  if (t - lastCheckTime < repartitionCheckInterval) {
    return false;
  }
  if (repartitionInterval < 0) {
    return false;
  }
  if (t - lastRepartitionTime < repartitionInterval) {
    return false;
  }
  kafkaStatsHolder.put(opid,kstats);
  if (kafkaStatsHolder.size() != currentPartitionInfo.size() || currentPartitionInfo.size() == 0) {
    return false;
  }
  try {
{
      Set<KafkaPartition> existingIds=new HashSet<KafkaPartition>();
      for (      PartitionInfo pio : currentPartitionInfo) {
        existingIds.addAll(pio.kpids);
      }
      Map<String,List<PartitionMetadata>> partitionsMeta=KafkaMetadataUtil.getPartitionsForTopic(consumer.brokers,consumer.getTopic());
      if (partitionsMeta == null) {
        return false;
      }
      for (      Map.Entry<String,List<PartitionMetadata>> en : partitionsMeta.entrySet()) {
        if (en.getValue() == null) {
          continue;
        }
        for (        PartitionMetadata pm : en.getValue()) {
          KafkaPartition pa=new KafkaPartition(en.getKey(),consumer.topic,pm.partitionId());
          if (!existingIds.contains(pa)) {
            newWaitingPartition.add(pa);
          }
        }
      }
      if (newWaitingPartition.size() != 0) {
        lastRepartitionTime=t;
        return true;
      }
    }
    return false;
  }
  finally {
    lastCheckTime=System.currentTimeMillis();
  }
}

private boolean isPartitionRequired ( int opid , List < KafkaConsumer . KafkaMeterStats > kstats ) { long t = System . currentTimeMillis ( ) ; if ( kstats . size ( ) > 0 ) { logger . debug ( "Checking offset updates for offset manager" ) ; updateOffsets ( kstats ) ; } if ( t - lastCheckTime < repartitionCheckInterval ) { return false ; } if ( repartitionInterval < 0 ) { return false ; } if ( t - lastRepartitionTime < repartitionInterval ) { return false ; } kafkaStatsHolder . put ( opid , kstats ) ; if ( kafkaStatsHolder . size ( ) != currentPartitionInfo . size ( ) || currentPartitionInfo . size ( ) == 0 ) { return false ; } try { { Set < KafkaPartition > existingIds = new HashSet < KafkaPartition > ( ) ; for ( PartitionInfo pio : currentPartitionInfo ) { existingIds . addAll ( pio . kpids ) ; } Map < String , List < PartitionMetadata > > partitionsMeta = KafkaMetadataUtil . getPartitionsForTopic ( consumer . brokers , consumer . getTopic ( ) ) ; if ( partitionsMeta == null ) { return false ; } for ( Map . Entry < String , List < PartitionMetadata > > en : partitionsMeta . entrySet ( ) ) { if ( en . getValue ( ) == null ) { continue ; } for ( PartitionMetadata pm : en . getValue ( ) ) { KafkaPartition pa = new KafkaPartition ( en . getKey ( ) , consumer . topic , pm . partitionId ( ) ) ; if ( ! existingIds . contains ( pa ) ) { newWaitingPartition . add ( pa ) ; } } } if ( newWaitingPartition . size ( ) != 0 ) { lastRepartitionTime = t ; return true ; } } return false ; } finally { lastCheckTime = System . currentTimeMillis ( ) ; } }
check whether the operator needs repartition based on reported stats
