public synchronized boolean safeCount(int partition,long offset) throws IOException {
  AtomicLong counter=null;
  if (partition >= this.counters.length) {
    this.counters=Arrays.copyOf(this.counters,this.counters.length + GROWBY);
  }
  counter=counters[partition];
  if (null == counter) {
    counter=new AtomicLong(0L);
    counters[partition]=counter;
    counter.set(offset);
    if (null != committedOffsets && committedOffsets.length > partition && committedOffsets[partition] >= 0 && 2 < (offset - committedOffsets[partition])) {
      labels.clear();
      labels.put(SensisionConstants.SENSISION_LABEL_TOPIC,this.topic);
      labels.put(SensisionConstants.SENSISION_LABEL_GROUPID,this.groupid);
      labels.put(SensisionConstants.SENSISION_LABEL_PARTITION,Integer.toString(partition));
      Sensision.update(SensisionConstants.SENSISION_CLASS_WARP_KAFKA_CONSUMER_OFFSET_FORWARD_LEAPS,labels,1);
      throw new IOException("Kafka offset " + offset + " leapt forward relative to previously committed value "+ committedOffsets[partition]);
    }
  }
 else {
    long previousOffset=counter.getAndSet(offset);
    if (1 < (offset - previousOffset)) {
      labels.clear();
      labels.put(SensisionConstants.SENSISION_LABEL_TOPIC,this.topic);
      labels.put(SensisionConstants.SENSISION_LABEL_GROUPID,this.groupid);
      labels.put(SensisionConstants.SENSISION_LABEL_PARTITION,Integer.toString(partition));
      Sensision.update(SensisionConstants.SENSISION_CLASS_WARP_KAFKA_CONSUMER_OFFSET_FORWARD_LEAPS,labels,1);
      throw new IOException("Kafka offset " + offset + " leapt forward relative to previous offset "+ previousOffset);
    }
  }
  if (null != committedOffsets && committedOffsets.length > partition && offset <= committedOffsets[partition]) {
    labels.clear();
    labels.put(SensisionConstants.SENSISION_LABEL_TOPIC,this.topic);
    labels.put(SensisionConstants.SENSISION_LABEL_GROUPID,this.groupid);
    labels.put(SensisionConstants.SENSISION_LABEL_PARTITION,Integer.toString(partition));
    Sensision.update(SensisionConstants.SENSISION_CLASS_WARP_KAFKA_CONSUMER_OFFSET_BACKWARD_LEAPS,labels,1);
    return false;
  }
  return true;
}

public synchronized boolean safeCount ( int partition , long offset ) throws IOException { AtomicLong counter = null ; if ( partition >= this . counters . length ) { this . counters = Arrays . copyOf ( this . counters , this . counters . length + GROWBY ) ; } counter = counters [ partition ] ; if ( null == counter ) { counter = new AtomicLong ( 0L ) ; counters [ partition ] = counter ; counter . set ( offset ) ; if ( null != committedOffsets && committedOffsets . length > partition && committedOffsets [ partition ] >= 0 && 2 < ( offset - committedOffsets [ partition ] ) ) { labels . clear ( ) ; labels . put ( SensisionConstants . SENSISION_LABEL_TOPIC , this . topic ) ; labels . put ( SensisionConstants . SENSISION_LABEL_GROUPID , this . groupid ) ; labels . put ( SensisionConstants . SENSISION_LABEL_PARTITION , Integer . toString ( partition ) ) ; Sensision . update ( SensisionConstants . SENSISION_CLASS_WARP_KAFKA_CONSUMER_OFFSET_FORWARD_LEAPS , labels , 1 ) ; throw new IOException ( "Kafka offset " + offset + " leapt forward relative to previously committed value " + committedOffsets [ partition ] ) ; } } else { long previousOffset = counter . getAndSet ( offset ) ; if ( 1 < ( offset - previousOffset ) ) { labels . clear ( ) ; labels . put ( SensisionConstants . SENSISION_LABEL_TOPIC , this . topic ) ; labels . put ( SensisionConstants . SENSISION_LABEL_GROUPID , this . groupid ) ; labels . put ( SensisionConstants . SENSISION_LABEL_PARTITION , Integer . toString ( partition ) ) ; Sensision . update ( SensisionConstants . SENSISION_CLASS_WARP_KAFKA_CONSUMER_OFFSET_FORWARD_LEAPS , labels , 1 ) ; throw new IOException ( "Kafka offset " + offset + " leapt forward relative to previous offset " + previousOffset ) ; } } if ( null != committedOffsets && committedOffsets . length > partition && offset <= committedOffsets [ partition ] ) { labels . clear ( ) ; labels . put ( SensisionConstants . SENSISION_LABEL_TOPIC , this . topic ) ; labels . put ( SensisionConstants . SENSISION_LABEL_GROUPID , this . groupid ) ; labels . put ( SensisionConstants . SENSISION_LABEL_PARTITION , Integer . toString ( partition ) ) ; Sensision . update ( SensisionConstants . SENSISION_CLASS_WARP_KAFKA_CONSUMER_OFFSET_BACKWARD_LEAPS , labels , 1 ) ; return false ; } return true ; }
will account for the new offset , checking several things : 1 . - the offset is not below the previously committed offset 2 . - the offset does not differ of more than 1 from the previously known counter if 1 . happens , safecount will return ' false ' . if 2 . happens , safecount will throw an ioexception .
