本文整理了Java中org.apache.kafka.clients.consumer.Consumer.committed()方法的一些代码示例，展示了Consumer.committed()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Consumer.committed()方法的具体详情如下：
包路径：org.apache.kafka.clients.consumer.Consumer
类名称：Consumer
方法名：committed

Consumer.committed介绍

[英]Fetches the last committed offsets for the input list of partitions
[中]获取分区输入列表的最后提交的偏移量

代码示例

代码示例来源：origin: openzipkin/brave

public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) {
 return delegate.committed(partition, timeout);
}

代码示例来源：origin: openzipkin/brave

@Override public OffsetAndMetadata committed(TopicPartition partition) {
 return delegate.committed(partition);
}

代码示例来源：origin: apache/nifi

private void rollback(final TopicPartition topicPartition) {
  OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition);
  if (offsetAndMetadata == null) {
    offsetAndMetadata = kafkaConsumer.committed(topicPartition);
  }
  final long offset = offsetAndMetadata.offset();
  kafkaConsumer.seek(topicPartition, offset);
}

代码示例来源：origin: apache/nifi

private void rollback(final TopicPartition topicPartition) {
  try {
    OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition);
    if (offsetAndMetadata == null) {
      offsetAndMetadata = kafkaConsumer.committed(topicPartition);
    }
    final long offset = offsetAndMetadata == null ? 0L : offsetAndMetadata.offset();
    kafkaConsumer.seek(topicPartition, offset);
  } catch (final Exception rollbackException) {
    logger.warn("Attempted to rollback Kafka message offset but was unable to do so", rollbackException);
  }
}

代码示例来源：origin: apache/nifi

private void rollback(final TopicPartition topicPartition) {
  try {
    OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition);
    if (offsetAndMetadata == null) {
      offsetAndMetadata = kafkaConsumer.committed(topicPartition);
    }
    final long offset = offsetAndMetadata == null ? 0L : offsetAndMetadata.offset();
    kafkaConsumer.seek(topicPartition, offset);
  } catch (final Exception rollbackException) {
    logger.warn("Attempted to rollback Kafka message offset but was unable to do so", rollbackException);
  }
}

代码示例来源：origin: apache/nifi

private void rollback(final TopicPartition topicPartition) {
  try {
    OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition);
    if (offsetAndMetadata == null) {
      offsetAndMetadata = kafkaConsumer.committed(topicPartition);
    }
    final long offset = offsetAndMetadata == null ? 0L : offsetAndMetadata.offset();
    kafkaConsumer.seek(topicPartition, offset);
  } catch (final Exception rollbackException) {
    logger.warn("Attempted to rollback Kafka message offset but was unable to do so", rollbackException);
  }
}

代码示例来源：origin: spring-projects/spring-kafka

@KafkaListener(id = "batchAckListener", topics = { "annotated26", "annotated27" },
    containerFactory = "batchFactory")
public void batchAckListener(List<String> in,
    @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
    @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
    Consumer<?, ?> consumer) {
  for (int i = 0; i < topics.size(); i++) {
    this.latch17.countDown();
    String topic = topics.get(i);
    if ("annotated26".equals(topic) && consumer.committed(
        new org.apache.kafka.common.TopicPartition(topic, partitions.get(i))).offset() == 1) {
      this.latch18.countDown();
    }
    else if ("annotated27".equals(topic) && consumer.committed(
        new org.apache.kafka.common.TopicPartition(topic, partitions.get(i))).offset() == 3) {
      this.latch18.countDown();
    }
  }
}

代码示例来源：origin: opentracing-contrib/java-kafka-client

@Override
public OffsetAndMetadata committed(TopicPartition partition) {
 return consumer.committed(partition);
}

代码示例来源：origin: opentracing-contrib/java-kafka-client

@Override
public OffsetAndMetadata committed(TopicPartition topicPartition, Duration duration) {
 return consumer.committed(topicPartition, duration);
}

代码示例来源：origin: io.opentracing.contrib/opentracing-kafka-client

@Override
public OffsetAndMetadata committed(TopicPartition topicPartition, Duration duration) {
 return consumer.committed(topicPartition, duration);
}

代码示例来源：origin: io.opentracing.contrib/opentracing-kafka-client

@Override
public OffsetAndMetadata committed(TopicPartition partition) {
 return consumer.committed(partition);
}

代码示例来源：origin: rayokota/kafka-graphs

@Override
public OffsetAndMetadata committed(TopicPartition partition) {
  return kafkaConsumer.committed(partition);
}

代码示例来源：origin: rayokota/kafka-graphs

@Override
public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) {
  return kafkaConsumer.committed(partition, timeout);
}

代码示例来源：origin: io.zipkin.brave/brave-instrumentation-kafka-clients

public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) {
 return delegate.committed(partition, timeout);
}

代码示例来源：origin: vert-x3/vertx-kafka-client

@Override
public void committed(TopicPartition topicPartition, Handler<AsyncResult<OffsetAndMetadata>> handler) {
 this.submitTask((consumer, future) -> {
  OffsetAndMetadata result = consumer.committed(topicPartition);
  if (future != null) {
   future.complete(result);
  }
 }, handler);
}

代码示例来源：origin: linkedin/li-apache-kafka-clients

@Override
public Long committedSafeOffset(TopicPartition tp) {
 OffsetAndMetadata rawOffsetAndMetadata = _kafkaConsumer.committed(tp);
 if (rawOffsetAndMetadata == null || rawOffsetAndMetadata.metadata().isEmpty()) {
  return null;
 }
 return rawOffsetAndMetadata.offset();
}

代码示例来源：origin: org.apache.nifi/nifi-kafka-0-10-processors

private void rollback(final TopicPartition topicPartition) {
  OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition);
  if (offsetAndMetadata == null) {
    offsetAndMetadata = kafkaConsumer.committed(topicPartition);
  }
  final long offset = offsetAndMetadata.offset();
  kafkaConsumer.seek(topicPartition, offset);
}

代码示例来源：origin: org.apache.nifi/nifi-kafka-2-0-processors

private void rollback(final TopicPartition topicPartition) {
  try {
    OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition);
    if (offsetAndMetadata == null) {
      offsetAndMetadata = kafkaConsumer.committed(topicPartition);
    }
    final long offset = offsetAndMetadata == null ? 0L : offsetAndMetadata.offset();
    kafkaConsumer.seek(topicPartition, offset);
  } catch (final Exception rollbackException) {
    logger.warn("Attempted to rollback Kafka message offset but was unable to do so", rollbackException);
  }
}

代码示例来源：origin: org.apache.nifi/nifi-kafka-1-0-processors

private void rollback(final TopicPartition topicPartition) {
  try {
    OffsetAndMetadata offsetAndMetadata = uncommittedOffsetsMap.get(topicPartition);
    if (offsetAndMetadata == null) {
      offsetAndMetadata = kafkaConsumer.committed(topicPartition);
    }
    final long offset = offsetAndMetadata == null ? 0L : offsetAndMetadata.offset();
    kafkaConsumer.seek(topicPartition, offset);
  } catch (final Exception rollbackException) {
    logger.warn("Attempted to rollback Kafka message offset but was unable to do so", rollbackException);
  }
}

代码示例来源：origin: reactor/reactor-kafka

private long committedCount(KafkaReceiver<Integer, String> receiver) {
  long committed = 0;
  for (int j = 0; j < partitions; j++) {
    TopicPartition p = new TopicPartition(topic, j);
    OffsetAndMetadata offset = receiver.doOnConsumer(c -> c.committed(p)).block(Duration.ofSeconds(receiveTimeoutMillis));
    if (offset != null && offset.offset() > 0)
      committed += offset.offset();
  }
  return committed;
}