org.apache.kafka.clients.producer.Producer.sendOffsetsToTransaction()方法的使用及代码示例

x33g5p2x  于2022-01-26 转载在 其他  
字(10.1k)|赞(0)|评价(0)|浏览(170)

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

Producer.sendOffsetsToTransaction介绍

[英]See KafkaProducer#sendOffsetsToTransaction(Map,String)
[中]参见KafkaProducer#sendOffsetsToTransaction（地图，字符串）

代码示例

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

@Override
 public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
   String consumerGroupId) {
  delegate.sendOffsetsToTransaction(offsets, consumerGroupId);
 }
}

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

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)
    throws ProducerFencedException {
  this.delegate.sendOffsetsToTransaction(offsets, consumerGroupId);
}

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

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) {
  @SuppressWarnings("unchecked")
  KafkaResourceHolder<K, V> resourceHolder = (KafkaResourceHolder<K, V>) TransactionSynchronizationManager
      .getResource(this.producerFactory);
  Assert.isTrue(resourceHolder != null, "No transaction in process");
  if (resourceHolder.getProducer() != null) {
    resourceHolder.getProducer().sendOffsetsToTransaction(offsets, consumerGroupId);
  }
}

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

Map<TopicPartition, OffsetAndMetadata> offsets = new LinkedHashMap<>();
offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(1L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(1L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 1L);
offsets.clear();
offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(1L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));

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

@SuppressWarnings("unchecked")
@Test
public void discardRemainingRecordsFromPollAndSeek() throws Exception {
  assertThat(this.config.deliveryLatch.await(10, TimeUnit.SECONDS)).isTrue();
  assertThat(this.config.pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
  this.registry.stop();
  assertThat(this.config.closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
  InOrder inOrder = inOrder(this.consumer, this.producer);
  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.producer).beginTransaction();
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 0), 0L);
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 0L);
  inOrder.verify(this.consumer).seek(new TopicPartition("foo", 2), 0L);
  inOrder.verify(this.producer).abortTransaction();
  inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
  inOrder.verify(this.producer).beginTransaction();
  Map<TopicPartition, OffsetAndMetadata> offsets = new LinkedHashMap<>();
  offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L));
  offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L));
  offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L));
  inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
  inOrder.verify(this.producer).commitTransaction();
  assertThat(this.config.ehException).isInstanceOf(ListenerExecutionFailedException.class);
  assertThat(((ListenerExecutionFailedException) this.config.ehException).getGroupId()).isEqualTo(CONTAINER_ID);
}

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

Map<TopicPartition, OffsetAndMetadata> offsets = new LinkedHashMap<>();
offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(1L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 0), new OffsetAndMetadata(2L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(1L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
inOrder.verify(this.consumer).seek(new TopicPartition("foo", 1), 1L);
offsets.clear();
offsets.put(new TopicPartition("foo", 1), new OffsetAndMetadata(2L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(1L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
offsets.clear();
offsets.put(new TopicPartition("foo", 2), new OffsetAndMetadata(2L));
inOrder.verify(this.producer).sendOffsetsToTransaction(offsets, CONTAINER_ID);
inOrder.verify(this.producer).commitTransaction();
inOrder.verify(this.consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));

代码示例来源：origin: org.axonframework/axon-kafka

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)
    throws ProducerFencedException {
  this.delegate.sendOffsetsToTransaction(offsets, consumerGroupId);
}

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

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                   String consumerGroupId) throws ProducerFencedException {
 _producer.sendOffsetsToTransaction(offsets, consumerGroupId);
}

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

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> map, String s)
  throws ProducerFencedException {
 producer.sendOffsetsToTransaction(map, s);
}

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

@Override
 public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
   String consumerGroupId) {
  delegate.sendOffsetsToTransaction(offsets, consumerGroupId);
 }
}

代码示例来源：origin: com.github.combinedmq/combinedmq

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException {
  producer.sendOffsetsToTransaction(offsets, consumerGroupId);
}

代码示例来源：origin: org.axonframework.extensions.kafka/axon-kafka

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId)
    throws ProducerFencedException {
  this.delegate.sendOffsetsToTransaction(offsets, consumerGroupId);
}

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

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> map, String s)
  throws ProducerFencedException {
 producer.sendOffsetsToTransaction(map, s);
}

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

verify(producer).send(captor.capture(), any(Callback.class));
assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
inOrder.verify(producer, never()).sendOffsetsToTransaction(anyMap(), anyString());
inOrder.verify(producer, never()).commitTransaction();
inOrder.verify(producer).abortTransaction();

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

InOrder inOrder = inOrder(producer);
inOrder.verify(producer).beginTransaction();
inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
    new OffsetAndMetadata(0)), "group");
inOrder.verify(producer).commitTransaction();
inOrder.verify(producer).send(captor.capture(), any(Callback.class));
assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
    new OffsetAndMetadata(1)), "group");
inOrder.verify(producer).commitTransaction();

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

verify(producer).send(captor.capture(), any(Callback.class));
assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
inOrder.verify(producer, never()).sendOffsetsToTransaction(anyMap(), anyString());
inOrder.verify(producer, never()).commitTransaction();
inOrder.verify(producer).abortTransaction();

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

inOrder.verify(producer).send(captor.capture(), any(Callback.class));
assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
    new OffsetAndMetadata(1)), "group");
inOrder.verify(producer).commitTransaction();

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

producer.sendOffsetsToTransaction(consumedOffsetsAndMetadata, applicationId);
producer.commitTransaction();
transactionInFlight = false;

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

Producer

相关文章

    查看更多

    热门标签

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

    最新文章

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