org.apache.kafka.clients.consumer.Consumer.unsubscribe()方法的使用及代码示例

x33g5p2x  于2022-01-18 转载在 其他  
字(9.0k)|赞(0)|评价(0)|浏览(172)

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

Consumer.unsubscribe介绍

暂无

代码示例

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

@Override public void unsubscribe() {
 delegate.unsubscribe();
}

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

private void closeConsumer(final Consumer<?, ?> consumer) {
  consumerClosedCountRef.incrementAndGet();
  try {
    consumer.unsubscribe();
  } catch (Exception e) {
    logger.warn("Failed while unsubscribing " + consumer, e);
  }
  try {
    consumer.close();
  } catch (Exception e) {
    logger.warn("Failed while closing " + consumer, e);
  }
}

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

private void closeConsumer(final Consumer<?, ?> consumer) {
  consumerClosedCountRef.incrementAndGet();
  try {
    consumer.unsubscribe();
  } catch (Exception e) {
    logger.warn("Failed while unsubscribing " + consumer, e);
  }
  try {
    consumer.close();
  } catch (Exception e) {
    logger.warn("Failed while closing " + consumer, e);
  }
}

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

private void closeConsumer(final Consumer<?, ?> consumer) {
  consumerClosedCountRef.incrementAndGet();
  try {
    consumer.unsubscribe();
  } catch (Exception e) {
    logger.warn("Failed while unsubscribing " + consumer, e);
  }
  try {
    consumer.close();
  } catch (Exception e) {
    logger.warn("Failed while closing " + consumer, e);
  }
}

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

private void closeConsumer(final Consumer<?, ?> consumer) {
  consumerClosedCountRef.incrementAndGet();
  try {
    consumer.unsubscribe();
  } catch (Exception e) {
    logger.warn("Failed while unsubscribing " + consumer, e);
  }
  try {
    consumer.close();
  } catch (Exception e) {
    logger.warn("Failed while closing " + consumer, e);
  }
}

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

private void closeConsumer(final Consumer<?, ?> consumer) {
  consumerClosedCountRef.incrementAndGet();
  try {
    consumer.unsubscribe();
  } catch (Exception e) {
    logger.warn("Failed while unsubscribing " + consumer, e);
  }
  try {
    consumer.close();
  } catch (Exception e) {
    logger.warn("Failed while closing " + consumer, e);
  }
}

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

@Override
public String getSampleMessage(final String topic) {
 String message = null;
 if (listTopics().contains(topic)) {
  try (Consumer<String, String> kafkaConsumer = kafkaConsumerFactory.createConsumer()) {
   kafkaConsumer.assign(kafkaConsumer.partitionsFor(topic).stream()
    .map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
    .collect(Collectors.toList()));
   kafkaConsumer.assignment().stream()
    .filter(p -> (kafkaConsumer.position(p) - 1) >= 0)
    .forEach(p -> kafkaConsumer.seek(p, kafkaConsumer.position(p) - 1));
   final ConsumerRecords<String, String> records = kafkaConsumer.poll(KAFKA_CONSUMER_TIMEOUT);
   message = records.isEmpty() ? null : records.iterator().next().value();
   kafkaConsumer.unsubscribe();
  }
 }
 return message;
}

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

@SuppressWarnings("unchecked")
@Test
public void stopContainerAfterException() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID);
  assertThat(container.isRunning()).isFalse();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.consumer).wakeup();
  inOrder.verify(this.consumer).unsubscribe();
  inOrder.verify(this.consumer).close();
  inOrder.verifyNoMoreInteractions();
}

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

@SuppressWarnings("unchecked")
@Test
public void stopContainerAfterException() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID);
  assertThat(container.isRunning()).isFalse();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.consumer).wakeup();
  inOrder.verify(this.consumer).unsubscribe();
  inOrder.verify(this.consumer).close();
  inOrder.verifyNoMoreInteractions();
  assertThat(this.config.count).isEqualTo(4);
  assertThat(this.config.contents.toArray()).isEqualTo(new String[]
      { "foo", "bar", "baz", "qux" });
}

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

@SuppressWarnings("unchecked")
@Test
public void stopContainerAfterException() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.errorLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  MessageListenerContainer container = this.registry.getListenerContainer(CONTAINER_ID);
  assertThat(container.isRunning()).isFalse();
  InOrder inOrder = inOrder(this.consumer);
  inOrder.verify(this.consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(1L)));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(2L)));
  inOrder.verify(this.consumer).commitSync(
      Collections.singletonMap(new TopicPartition("foo", 1), new OffsetAndMetadata(1L)));
  inOrder.verify(this.consumer).wakeup();
  inOrder.verify(this.consumer).unsubscribe();
  inOrder.verify(this.consumer).close();
  inOrder.verifyNoMoreInteractions();
  assertThat(this.config.count).isEqualTo(4);
  assertThat(this.config.contents.toArray()).isEqualTo(new String[]
      { "foo", "bar", "baz", "qux" });
}

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

@Override
public void unsubscribe() {
  kafkaConsumer.unsubscribe();
}

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

@Override
public KafkaReadStream<K, V> unsubscribe(Handler<AsyncResult<Void>> completionHandler) {
 this.submitTask((consumer, future) -> {
  consumer.unsubscribe();
  if (future != null) {
   future.complete();
  }
 }, completionHandler);
 return this;
}

代码示例来源：origin: com.hotels.road/road-offramp-kafka

@Override
public void close() {
 try {
  consumer.unsubscribe();
  consumer.close();
  log.info("Closed KafkaConsumer");
 } catch (org.apache.kafka.common.errors.InterruptException e) {
  log.error("Interrupted while closing KafkaConsumer");
  Thread.currentThread().interrupt();
 }
}

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

@Override
public void unsubscribe() {
 // Clear all the state of the topic in consumer record processor.
 _consumerRecordsProcessor.clear();
 _kafkaConsumer.unsubscribe();
}

代码示例来源：origin: authorjapps/zerocode

public static void handleSeekOffset(ConsumerLocalConfigs effectiveLocal, Consumer consumer) {
  String seek = effectiveLocal.getSeek();
  if (!isEmpty(seek)) {
    String[] seekPosition = effectiveLocal.getSeekTopicPartitionOffset();
    TopicPartition topicPartition = new TopicPartition(seekPosition[0], parseInt(seekPosition[1]));
    Set<TopicPartition> topicPartitions = new HashSet<>();
    topicPartitions.add(topicPartition);
    consumer.unsubscribe();
    consumer.assign(topicPartitions);
    consumer.seek(topicPartition, parseLong(seekPosition[2]));
  }
}

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

@Override
  public void close() {
    super.close();
    currentRow = null;
    consumer.unsubscribe();
    consumer.close();
  }
}

代码示例来源：origin: org.apache.metamodel/MetaModel-kafka

@Override
  public void close() {
    super.close();
    currentRow = null;
    consumer.unsubscribe();
    consumer.close();
  }
}

代码示例来源：origin: org.apache.kafka/kafka-streams

private void enforceRebalance() {
  consumer.unsubscribe();
  consumer.subscribe(builder.sourceTopicPattern(), rebalanceListener);
}

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

private void closeConsumer(final Consumer<?, ?> consumer) {
  consumerClosedCountRef.incrementAndGet();
  try {
    consumer.unsubscribe();
  } catch (Exception e) {
    logger.warn("Failed while unsubscribing " + consumer, e);
  }
  try {
    consumer.close();
  } catch (Exception e) {
    logger.warn("Failed while closing " + consumer, e);
  }
}

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

private void closeConsumer(final Consumer<?, ?> consumer) {
  consumerClosedCountRef.incrementAndGet();
  try {
    consumer.unsubscribe();
  } catch (Exception e) {
    logger.warn("Failed while unsubscribing " + consumer, e);
  }
  try {
    consumer.close();
  } catch (Exception e) {
    logger.warn("Failed while closing " + consumer, e);
  }
}

内容来源于网络，如有侵权，请联系作者删除！

Consumer

相关文章

    查看更多
    微信公众号

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com