Compares the current consumer offsets vs the its previous state Returns topic payload that wasn't incremented from last check
546127
function getNotIncrementedTopicPayloads(previousConsumerReadOffset, consumer) {
    let notIncrementedTopicPayloads = consumer.topicPayloads.filter((topicPayload) => {
        let {topic, partition, offset: currentOffset} = topicPayload;
        let previousTopicPayloadForPartition = _.find(previousConsumerReadOffset, {topic, partition});
        return previousTopicPayloadForPartition && currentOffset === previousTopicPayloadForPartition.offset;
    });

    return notIncrementedTopicPayloads;
}
1029155
public boolean compareAndSet(T expectedState, T newState)
    {
        checkState(!Thread.holdsLock(lock), "Can not set state while holding the lock");
        requireNonNull(expectedState, "expectedState is null");
        requireNonNull(newState, "newState is null");

        FutureStateChange<T> futureStateChange;
        ImmutableList<StateChangeListener<T>> stateChangeListeners;
        synchronized (lock) {
            if (!state.equals(expectedState)) {
                return false;
            }

            // change to same state is not a change, and does not notify the notify listeners
            if (state.equals(newState)) {
                return false;
            }

            checkState(!isTerminalState(state), "%s can not transition from %s to %s", name, state, newState);

            state = newState;

            futureStateChange = this.futureStateChange.getAndSet(new FutureStateChange<>());
            stateChangeListeners = ImmutableList.copyOf(this.stateChangeListeners);

            // if we are now in a terminal state, free the listeners since this will be the last notification
            if (isTerminalState(state)) {
                this.stateChangeListeners.clear();
            }
        }

        fireStateChanged(newState, futureStateChange, stateChangeListeners);
        return true;
    }
1039343
public static boolean seek(KafkaConsumer<?, ?> consumer, KafkaTopicPartitionOffset tpo) {
        synchronized (consumer) {
            Set<TopicPartition> topicParts = consumer.assignment();
            if (topicParts != null) {
                for (TopicPartition tp : topicParts) {
                    if (StringUtils.equals(tpo.topic, tp.topic())
                            && tpo.partition == tp.partition()) {
                        consumer.seek(tp, tpo.offset);
                        // we want to seek as soon as possible
                        // since seekToEnd evaluates lazily, invoke position()
                        // so
                        // that seeking will be committed.
                        consumer.position(tp);
                        return true;
                    }
                }
            }
        }
        return false;
    }
1235250
function hierarchicalTopicDispatcher (state, topic, data, options) {
    var
      lineage = findLineage(getTopic, isAncestorTopic, topic, state._topics),
      topicNode = lineage[lineage.length - 1],
      subscriptions = options.preventBubble
        ? topicNode.topic === topic ? topicNode.subscriptions : []
        : mergeBy(getFingerArrayPriority, map(getSubscriptions, lineage)),
      fulfilledPromise = subscriptionDispatcher(
        topic, data, options, subscriptions
      );

    if (options.persist) {
      var id = state.id();

      topicNode = addTopicLine(topic, topicNode);
      topicNode.persisted.push(
        {topic: topic, data: data, order: id}
      );
      fulfilledPromise.token = {
        topic: topic,
        id: id
      };
    }

    return fulfilledPromise;
  }
1179175
private KafkaConsumer<String, byte[]> _getConsumer(String topic, boolean autoCommitOffsets) {
        KafkaConsumer<String, byte[]> consumer = topicConsumers.get(topic);
        if (consumer == null) {
            consumer = KafkaHelper.createKafkaConsumer(bootstrapServers, consumerGroupId,
                    consumeFromBeginning, autoCommitOffsets, consumerProperties);
            KafkaConsumer<String, byte[]> existingConsumer = topicConsumers.putIfAbsent(topic,
                    consumer);
            if (existingConsumer != null) {
                consumer.close();
                consumer = existingConsumer;
            } else {
                _checkAndSubscribe(consumer, topic);
            }
        }
        return consumer;
    }
570648
def update_consumer_offsets(self, partition_offsets):
        
        self.logger.debug("Updating consumer offsets to: %s", partition_offsets)

        for partition, offset in partition_offsets.items():
            self.consumer.offsets[partition] = offset

        # consumer keeps other offset states beyond its `offsets` dictionary,
        # a relative seek with zero delta forces the consumer to reset to the
        # current value of the `offsets` dictionary
        self.consumer.seek(0, 1)
1205335
def _find_topics_with_wrong_rp(topics, zk, default_min_isr):
    """"""
    topics_with_wrong_rf = []

    for topic_name, partitions in topics.items():
        min_isr = get_min_isr(zk, topic_name) or default_min_isr
        replication_factor = len(partitions[0].replicas)

        if replication_factor >= min_isr + 1:
            continue

        topics_with_wrong_rf.append({
            'replication_factor': replication_factor,
            'min_isr': min_isr,
            'topic': topic_name,
        })

    return topics_with_wrong_rf
559839
def parse_consumer_offsets(cls, json_file):
        """"""
        with open(json_file, 'r') as consumer_offsets_json:
            try:
                parsed_offsets = {}
                parsed_offsets_data = json.load(consumer_offsets_json)
                # Create new dict with partition-keys as integers
                parsed_offsets['groupid'] = parsed_offsets_data['groupid']
                parsed_offsets['offsets'] = {}
                for topic, topic_data in six.iteritems(parsed_offsets_data['offsets']):
                    parsed_offsets['offsets'][topic] = {}
                    for partition, offset in six.iteritems(topic_data):
                        parsed_offsets['offsets'][topic][int(partition)] = offset
                return parsed_offsets
            except ValueError:
                print(
                    "Error: Given consumer-data json data-file {file} could not be "
                    "parsed".format(file=json_file),
                    file=sys.stderr,
                )
                raise
1285109
@Override
    public E poll() {
        // local load of field to avoid repeated loads after volatile reads
        final long[] sBuffer = sequenceBuffer;
        final long mask = this.mask;

        long cIndex;
        long seq;
        long seqOffset;
        long expectedSeq;
        long pIndex = -1; // start with bogus value, hope we don't need it
        do {
            cIndex = lvConsumerIndex();
            seqOffset = calcSequenceOffset(cIndex, mask);
            seq = lvSequence(sBuffer, seqOffset);
            expectedSeq = cIndex + 1;
            if (seq < expectedSeq) { // slot has not been moved by producer
                if (cIndex >= pIndex && // test against cached pIndex
                    cIndex == (pIndex = lvProducerIndex())) { // update pIndex if we must
                    // strict empty check, this ensures [Queue.poll() == null iff isEmpty()]
                    return null;
                } else {
                    seq = expectedSeq + 1; // trip another go around
                }
            }
        } while (seq > expectedSeq || // another consumer beat us to it
                !casConsumerIndex(cIndex, cIndex + 1)); // failed the CAS

        final long offset = calcElementOffset(cIndex, mask);
        final E e = lpElement(buffer, offset);
        assert e != null;
        soElement(buffer, offset, null);
        soSequence(sBuffer, seqOffset, cIndex + mask + 1);// i.e. seq += capacity
        return e;
    }
771449
function(topic, cb) {
    if (topic.params.offset === undefined || topic.params.offset === -1) {
      var fetchFrom = topic.broker;

      if (topic.params.fetchFromCoordinator) {
        fetchFrom = _groupCoordinator;
      }

      fetchFrom.offsetFetch({
        group: _options.group,
        topic: topic.params.topic,
        partition: topic.params.partition,
        fetchFromCoordinator: topic.params.fetchFromCoordinator
      }, function(err, response) {
        if (err) { return cb(err); }
        var partition = _.findWhere(_.findWhere(response.topics, {name: topic.params.topic}).partitions, {partitionId: topic.params.partition});
        if (partition.error === 0) {
          if (partition.offsetLo === 0xffffffff &&
              partition.offsetHi === 0xffffffff) {

            // BUG in Kafka - if the topic offset has not been set then an error condition is returned and
            // -1 is set for the offset, once an offset is set this condition is cleared
            topic.params.offset = 0;
          } else {
            topic.params.offset = partition.offsetLo;
          }
          cb(null, topic);
        }
        else {
          cb(partition.error, null);
        }
      });
    }
    else {
      cb(null, topic);
    }
  }
957765
def seek(self, offset, topic_partition=None):
        

        if isinstance(topic_partition, TopicPartition):
            self.consumer.seek(topic_partition, offset=offset)
        else:
            raise TypeError("topic_partition must be of type TopicPartition, create it with Create TopicPartition keyword.")
225598
public int consume(IndexValueConsumer consumer) {
        int filledChunks = getFilledChunks();
        int offset;
        for (int i = 0; i < filledChunks ; i++) {
            offset = i * chunkSize;
            int[] chunk = chunks[i];
            if (chunk!=null) {
                for (int j = 0; j < chunkSize; j++) {
                    consumer.accept(offset+j, chunk[i]);
                }
            }
        }
        int remainder = size() % chunkSize;
        if (remainder != 0) {
            offset = filledChunks * chunkSize;
            int[] chunk = chunks[filledChunks];
            for (int j = 0; j < remainder; j++) {
                consumer.accept(offset+j, chunk[j]);
            }
        }
        return size();
    }
24242
private T pollInternal(boolean peek) {
        this.writeLock.lock();
        try {
            // Re-check size within the write lock
            if (this.size == 0) {
                return null;
            }

            final K key;
            if (this.peekedKey != null) {
                // If there is a peeked key use it
                key = this.peekedKey;
                if (!peek) {
                    // If not a peek consume the peekedKey
                    this.peekedKey = null;
                }
            } else {
                // Get the next element key
                key = this.getNextElementKey();
                if (peek) {
                    // If a peek store the key
                    this.peekedKey = key;
                }
            }

            // Get the associated Queue and sanitity check the value from getNextElementKey()
            final Queue<T> queue = this.keyedQueues.get(key);
            if (queue == null || queue.isEmpty()) {
                throw new IllegalStateException(
                        "getNextElementKey returned key='"
                                + key
                                + "' but there are no elements available for the key. This violates the contract specified for getNextElementKey");
            }

            if (peek) {
                // If a peek just return a peek from the queue
                return queue.peek();
            }

            // Not a peek, decrement the size and poll the queue
            this.size--;
            this.notFull.signal();
            return queue.poll();
        } finally {
            this.writeLock.unlock();
        }
    }
217115
def next_offsets(self):
        # type: (Descriptor) -> Offsets
        
        resume_bytes = self._resume()
        with self._meta_lock:
            if self._chunk_num >= self._total_chunks:
                return None, resume_bytes
            if self._offset + self._chunk_size > self._ase.size:
                num_bytes = self._ase.size - self._offset
            else:
                num_bytes = self._chunk_size
            chunk_num = self._chunk_num
            range_start = self._offset
            range_end = self._offset + num_bytes - 1
            self._offset += num_bytes
            self._chunk_num += 1
            if self._ase.is_encrypted and self._offset >= self._ase.size:
                pad = True
            else:
                pad = False
            return Offsets(
                chunk_num=chunk_num,
                num_bytes=num_bytes,
                range_start=range_start,
                range_end=range_end,
                pad=pad,
            ), resume_bytes
270147
private synchronized void getAllPreviousOffsetState(SourceState state) {
    if (this.doneGettingAllPreviousOffsets) {
      return;
    }
    this.previousOffsets.clear();
    this.previousLowWatermarks.clear();
    this.previousExpectedHighWatermarks.clear();
    this.previousOffsetFetchEpochTimes.clear();
    this.previousStartFetchEpochTimes.clear();
    this.previousStopFetchEpochTimes.clear();
    Map<String, Iterable<WorkUnitState>> workUnitStatesByDatasetUrns = state.getPreviousWorkUnitStatesByDatasetUrns();

    if (!workUnitStatesByDatasetUrns.isEmpty() &&
            !(workUnitStatesByDatasetUrns.size() == 1 && workUnitStatesByDatasetUrns.keySet().iterator().next()
        .equals(""))) {
      this.isDatasetStateEnabled.set(true);
    }

    for (WorkUnitState workUnitState : state.getPreviousWorkUnitStates()) {
      List<KafkaPartition> partitions = KafkaUtils.getPartitions(workUnitState);
      WorkUnit workUnit = workUnitState.getWorkunit();

      MultiLongWatermark watermark = workUnitState.getActualHighWatermark(MultiLongWatermark.class);
      MultiLongWatermark previousLowWatermark = workUnit.getLowWatermark(MultiLongWatermark.class);
      MultiLongWatermark previousExpectedHighWatermark = workUnit.getExpectedHighWatermark(MultiLongWatermark.class);
      Preconditions.checkArgument(partitions.size() == watermark.size(), String
          .format("Num of partitions doesn't match number of watermarks: partitions=%s, watermarks=%s", partitions,
              watermark));

      for (int i = 0; i < partitions.size(); i++) {
        KafkaPartition partition = partitions.get(i);

        if (watermark.get(i) != ConfigurationKeys.DEFAULT_WATERMARK_VALUE) {
          this.previousOffsets.put(partition, watermark.get(i));
        }

        if (previousLowWatermark.get(i) != ConfigurationKeys.DEFAULT_WATERMARK_VALUE) {
          this.previousLowWatermarks.put(partition, previousLowWatermark.get(i));
        }

        if (previousExpectedHighWatermark.get(i) != ConfigurationKeys.DEFAULT_WATERMARK_VALUE) {
          this.previousExpectedHighWatermarks.put(partition, previousExpectedHighWatermark.get(i));
        }

        this.previousOffsetFetchEpochTimes.put(partition,
            KafkaUtils.getPropAsLongFromSingleOrMultiWorkUnitState(workUnitState, OFFSET_FETCH_EPOCH_TIME, i));

        this.previousStartFetchEpochTimes.put(partition,
            KafkaUtils.getPropAsLongFromSingleOrMultiWorkUnitState(workUnitState, START_FETCH_EPOCH_TIME, i));

        this.previousStopFetchEpochTimes.put(partition,
            KafkaUtils.getPropAsLongFromSingleOrMultiWorkUnitState(workUnitState, STOP_FETCH_EPOCH_TIME, i));
      }
    }

    this.doneGettingAllPreviousOffsets = true;
  }
1072478
private void getOffset(long time, boolean local, int NonExistingTimeOpt, int DuplicatedTimeOpt, int[] offsets) {
        complete();
        TimeZoneRule rule = null;
        if (historicTransitions == null) {
            rule = initialRule;
        } else {
            long tstart = getTransitionTime(historicTransitions.get(0),
                    local, NonExistingTimeOpt, DuplicatedTimeOpt);
            if (time < tstart) {
                rule = initialRule;
            } else {
                int idx = historicTransitions.size() - 1;
                long tend = getTransitionTime(historicTransitions.get(idx),
                        local, NonExistingTimeOpt, DuplicatedTimeOpt);
                if (time > tend) {
                    if (finalRules != null) {
                        rule = findRuleInFinal(time, local, NonExistingTimeOpt, DuplicatedTimeOpt);
                    }
                    if (rule == null) {
                        // no final rules or the given time is before the first transition
                        // specified by the final rules -> use the last rule
                        rule = (historicTransitions.get(idx)).getTo();
                    }
                } else {
                    // Find a historical transition
                    while (idx >= 0) {
                        if (time >= getTransitionTime(historicTransitions.get(idx),
                                local, NonExistingTimeOpt, DuplicatedTimeOpt)) {
                            break;
                        }
                        idx--;
                    }
                    rule = (historicTransitions.get(idx)).getTo();
                }
            }
        }
        offsets[0] = rule.getRawOffset();
        offsets[1] = rule.getDSTSavings();
    }
617509
def pending(self, partitions=None):
        
        if partitions is None:
            partitions = self.offsets.keys()

        total = 0
        reqs = []

        for partition in partitions:
            reqs.append(OffsetRequestPayload(self.topic, partition, -1, 1))

        resps = self.client.send_offset_request(reqs)
        for resp in resps:
            partition = resp.partition
            pending = resp.offsets[0]
            offset = self.offsets[partition]
            total += pending - offset

        return total