Skip to content

KAFKA-7140: Remove deprecated poll usages #5319

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 11 commits into from
Aug 11, 2018
Merged

KAFKA-7140: Remove deprecated poll usages #5319

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
910317a
MINOR: use poll(Duration) in consumer example
viktorsomogyi Jul 2, 2018
fe77b3c
Replace other non-test usages as well
viktorsomogyi Jul 5, 2018
ab31aea
Removing the rest of the deprecated non-test poll usages
viktorsomogyi Jul 9, 2018
a359c53
Use empty iterator in ConsoleConsumer and manual assignment in EndToE…
viktorsomogyi Aug 9, 2018
87a57b7
Replace poll calls in WorkerSinkTaskTest
viktorsomogyi Aug 9, 2018
68c6197
Replace polls in ErrorHandlingTaskTest
viktorsomogyi Aug 9, 2018
6ad6f5e
Manual partition assignment in StreamsResetter
viktorsomogyi Aug 9, 2018
3482e73
Use partitionsFor in StreamsResetter
viktorsomogyi Aug 9, 2018
608624e
Remove unused imports
viktorsomogyi Aug 10, 2018
131b422
Auto create topic in EndToEndLatency
viktorsomogyi Aug 10, 2018
581292f
Fix failing tests in WorkerSinkTaskThreadedTest
viktorsomogyi Aug 10, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand Down Expand Up @@ -441,7 +442,7 @@ public String toString() {
}

private ConsumerRecords<byte[], byte[]> pollConsumer(long timeoutMs) {
ConsumerRecords<byte[], byte[]> msgs = consumer.poll(timeoutMs);
ConsumerRecords<byte[], byte[]> msgs = consumer.poll(Duration.ofMillis(timeoutMs));

// Exceptions raised from the task during a rebalance should be rethrown to stop the worker
if (rebalanceException != null) {
Expand Down
3 changes: 2 additions & 1 deletion connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Iterator;
Expand Down Expand Up @@ -253,7 +254,7 @@ private Consumer<K, V> createConsumer() {

private void poll(long timeoutMs) {
try {
ConsumerRecords<K, V> records = consumer.poll(timeoutMs);
ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(timeoutMs));
for (ConsumerRecord<K, V> record : records)
consumedCallback.onCompletion(null, record);
} catch (WakeupException e) {
Expand Down
5 changes: 3 additions & 2 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
Expand Down Expand Up @@ -180,8 +181,8 @@ public void testErrorHandlingInSinkTasks() throws Exception {
// bad json
ConsumerRecord<byte[], byte[]> record2 = new ConsumerRecord<>(TOPIC, PARTITION2, FIRST_OFFSET, null, "{\"a\" 10}".getBytes());

EasyMock.expect(consumer.poll(EasyMock.anyLong())).andReturn(records(record1));
EasyMock.expect(consumer.poll(EasyMock.anyLong())).andReturn(records(record2));
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andReturn(records(record1));
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andReturn(records(record2));

sinkTask.put(EasyMock.anyObject());
EasyMock.expectLastCall().times(2);
Expand Down
15 changes: 8 additions & 7 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.reflect.Whitebox;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
Expand Down Expand Up @@ -458,7 +459,7 @@ public void testWakeupInCommitSyncCausesRetry() throws Exception {
sinkTask.open(partitions);
EasyMock.expectLastCall();

EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andAnswer(
new IAnswer<ConsumerRecords<byte[], byte[]>>() {
@Override
public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
Expand Down Expand Up @@ -893,7 +894,7 @@ public void run() {
// Expect the next poll to discover and perform the rebalance, THEN complete the previous callback handler,
// and then return one record for TP1 and one for TP3.
final AtomicBoolean rebalanced = new AtomicBoolean();
EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andAnswer(
new IAnswer<ConsumerRecords<byte[], byte[]>>() {
@Override
public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
Expand Down Expand Up @@ -1273,7 +1274,7 @@ private void expectRebalanceRevocationError(RuntimeException e) {
sinkTask.preCommit(EasyMock.<Map<TopicPartition, OffsetAndMetadata>>anyObject());
EasyMock.expectLastCall().andReturn(Collections.emptyMap());

EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andAnswer(
new IAnswer<ConsumerRecords<byte[], byte[]>>() {
@Override
public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
Expand All @@ -1298,7 +1299,7 @@ private void expectRebalanceAssignmentError(RuntimeException e) {
sinkTask.open(partitions);
EasyMock.expectLastCall().andThrow(e);

EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andAnswer(
new IAnswer<ConsumerRecords<byte[], byte[]>>() {
@Override
public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
Expand All @@ -1315,7 +1316,7 @@ private void expectPollInitialAssignment() {
sinkTask.open(partitions);
EasyMock.expectLastCall();

EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(new IAnswer<ConsumerRecords<byte[], byte[]>>() {
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andAnswer(new IAnswer<ConsumerRecords<byte[], byte[]>>() {
@Override
public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
rebalanceListener.getValue().onPartitionsAssigned(partitions);
Expand All @@ -1332,15 +1333,15 @@ public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
private void expectConsumerWakeup() {
consumer.wakeup();
EasyMock.expectLastCall();
EasyMock.expect(consumer.poll(EasyMock.anyLong())).andThrow(new WakeupException());
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andThrow(new WakeupException());
}

private void expectConsumerPoll(final int numMessages) {
expectConsumerPoll(numMessages, RecordBatch.NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE);
}

private void expectConsumerPoll(final int numMessages, final long timestamp, final TimestampType timestampType) {
EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(
EasyMock.expect(consumer.poll(Duration.ofMillis(EasyMock.anyLong()))).andAnswer(
new IAnswer<ConsumerRecords<byte[], byte[]>>() {
@Override
public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
Expand Down
5 changes: 3 additions & 2 deletions core/src/main/scala/kafka/tools/ConsoleConsumer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package kafka.tools

import java.io.PrintStream
import java.nio.charset.StandardCharsets
import java.time.Duration
import java.util.concurrent.CountDownLatch
import java.util.regex.Pattern
import java.util.{Collections, Locale, Properties, Random}
Expand Down Expand Up @@ -388,7 +389,7 @@ object ConsoleConsumer extends Logging {
private[tools] class ConsumerWrapper(topic: Option[String], partitionId: Option[Int], offset: Option[Long], whitelist: Option[String],
consumer: Consumer[Array[Byte], Array[Byte]], val timeoutMs: Long = Long.MaxValue) {
consumerInit()
var recordIter = consumer.poll(0).iterator
var recordIter = Collections.emptyList[ConsumerRecord[Array[Byte], Array[Byte]]]().iterator()

def consumerInit() {
(topic, partitionId, offset, whitelist) match {
Expand Down Expand Up @@ -432,7 +433,7 @@ object ConsoleConsumer extends Logging {

def receive(): ConsumerRecord[Array[Byte], Array[Byte]] = {
if (!recordIter.hasNext) {
recordIter = consumer.poll(timeoutMs).iterator
recordIter = consumer.poll(Duration.ofMillis(timeoutMs)).iterator
if (!recordIter.hasNext)
throw new TimeoutException()
}
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/kafka/tools/ConsumerPerformance.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@ import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.{Metric, MetricName, TopicPartition}
import kafka.utils.{CommandLineUtils, ToolsUtils}
import java.util.{Collections, Properties, Random}

import java.text.SimpleDateFormat
import java.time.Duration

import com.typesafe.scalalogging.LazyLogging

Expand Down Expand Up @@ -127,7 +127,7 @@ object ConsumerPerformance extends LazyLogging {
var currentTimeMillis = lastConsumedTime

while (messagesRead < count && currentTimeMillis - lastConsumedTime <= timeout) {
val records = consumer.poll(100).asScala
val records = consumer.poll(Duration.ofMillis(100)).asScala
currentTimeMillis = System.currentTimeMillis
if (records.nonEmpty)
lastConsumedTime = currentTimeMillis
Expand Down
23 changes: 16 additions & 7 deletions core/src/main/scala/kafka/tools/EndToEndLatency.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,14 @@
package kafka.tools

import java.nio.charset.StandardCharsets
import java.time.Duration
import java.util
import java.util.{Arrays, Collections, Properties}

import kafka.utils.Exit
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import org.apache.kafka.clients.producer._
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.utils.Utils

import scala.collection.JavaConverters._
Expand Down Expand Up @@ -69,9 +72,7 @@ object EndToEndLatency {
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer")
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer")
consumerProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "0") //ensure we have no temporal batching

val consumer = new KafkaConsumer[Array[Byte], Array[Byte]](consumerProps)
consumer.subscribe(Collections.singletonList(topic))

val producerProps = loadProps
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
Expand All @@ -88,10 +89,18 @@ object EndToEndLatency {
consumer.close()
}

//Ensure we are at latest offset. seekToEnd evaluates lazily, that is to say actually performs the seek only when
//a poll() or position() request is issued. Hence we need to poll after we seek to ensure we see our first write.
consumer.seekToEnd(Collections.emptyList())
consumer.poll(0)
val topics = consumer.listTopics()

@viktorsomogyi viktorsomogyi Aug 9, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So here it seems the old code assumed that there is a topic. Now I added a check that there must be a topic beforehand. the producer would anyway create it according to the default settings but I think that most of the latency tests executed by users won't depend on the default settings and they rather want to create topics manually according to their likings.

@hachikuji hachikuji Aug 10, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess another approach might be to produce a dummy record to the topic first before assigning the partitions to the consumer and starting the test. That would let us retain the existing behavior.

val tp = topics.get(topic)
if (tp == null) {
finalise()
throw new RuntimeException("The tested topic doesn't exist in the cluster")
}
val topicPartitions = tp.asScala
.map(pi => new TopicPartition(pi.topic(), pi.partition()))
.to[List].asJava
consumer.assign(topicPartitions)
consumer.seekToEnd(topicPartitions)
consumer.assignment().asScala.foreach(consumer.position)

@hachikuji hachikuji Aug 8, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As far as I can tell, the assignment will be empty here. I don't think the old code worked either as intended because it did the seekToEnd prior to the call to poll(). For an end-to-end latency test, I actually wonder if we should just use manual assignment and take the group management out of the picture?

@viktorsomogyi viktorsomogyi Aug 9, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will change to manual assignment, something like this (which seems to work).

val topicPartitions = consumer.listTopics().get(topic).asScala
      .map(pi => new TopicPartition(pi.topic(), pi.partition()))
      .to[List].asJava
    consumer.assign(topicPartitions)
    consumer.seekToEnd(topicPartitions)
    consumer.assignment().asScala.foreach(consumer.position)


var totalTime = 0.0
val latencies = new Array[Long](numMessages)
Expand All @@ -103,7 +112,7 @@ object EndToEndLatency {

//Send message (of random bytes) synchronously then immediately poll for it
producer.send(new ProducerRecord[Array[Byte], Array[Byte]](topic, message)).get()
val recordIter = consumer.poll(timeout).iterator
val recordIter = consumer.poll(Duration.ofMillis(timeout)).iterator

val elapsed = System.nanoTime - begin

Expand Down
3 changes: 2 additions & 1 deletion core/src/main/scala/kafka/tools/MirrorMaker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package kafka.tools

import java.time.Duration
import java.util
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import java.util.concurrent.{CountDownLatch, TimeUnit}
Expand Down Expand Up @@ -452,7 +453,7 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
// uncommitted record since last poll. Using one second as poll's timeout ensures that
// offsetCommitIntervalMs, of value greater than 1 second, does not see delays in offset
// commit.
recordIter = consumer.poll(1000).iterator
recordIter = consumer.poll(Duration.ofSeconds(1)).iterator
if (!recordIter.hasNext)
throw new NoRecordsException
}
Expand Down
13 changes: 10 additions & 3 deletions core/src/main/scala/kafka/tools/StreamsResetter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
Expand All @@ -47,6 +48,7 @@
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
Expand All @@ -57,6 +59,7 @@
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
* {@link StreamsResetter} resets the processing state of a Kafka Streams application so that, for example, you can reprocess its input from scratch.
Expand Down Expand Up @@ -313,10 +316,14 @@ private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consum
config.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

try (final KafkaConsumer<byte[], byte[]> client = new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {
client.subscribe(topicsToSubscribe);
client.poll(1);
Map<String, List<PartitionInfo>> pi = client.listTopics();

@viktorsomogyi viktorsomogyi Aug 9, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not final yet, will refactor it a bit as suggested in earlier comments.

Collection<TopicPartition> partitions = pi.entrySet().stream()
.filter(entry -> topicsToSubscribe.contains(entry.getKey()))
.flatMap(entry -> entry.getValue().stream())
.map(info -> new TopicPartition(info.topic(), info.partition()))
.collect(Collectors.toList());
client.assign(partitions);

final Set<TopicPartition> partitions = client.assignment();
final Set<TopicPartition> inputTopicPartitions = new HashSet<>();
final Set<TopicPartition> intermediateTopicPartitions = new HashSet<>();

Expand Down
3 changes: 2 additions & 1 deletion examples/src/main/java/kafka/examples/Consumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

Expand All @@ -47,7 +48,7 @@ public Consumer(String topic) {
@Override
public void doWork() {
consumer.subscribe(Collections.singletonList(this.topic));
ConsumerRecords<Integer, String> records = consumer.poll(1000);
ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<Integer, String> record : records) {
System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
}
Expand Down
3 changes: 2 additions & 1 deletion tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.kafka.common.errors.ProducerFencedException;

import java.io.IOException;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
Expand Down Expand Up @@ -287,7 +288,7 @@ public void run() {
try {
producer.beginTransaction();
while (messagesInCurrentTransaction < numMessagesForNextTransaction) {
ConsumerRecords<String, String> records = consumer.poll(200L);
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
for (ConsumerRecord<String, String> record : records) {
producer.send(producerRecordFromConsumerRecord(outputTopic, record));
messagesInCurrentTransaction++;
Expand Down
3 changes: 2 additions & 1 deletion tools/src/main/java/org/apache/kafka/tools/VerifiableConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import java.io.Closeable;
import java.io.IOException;
import java.io.PrintStream;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
Expand Down Expand Up @@ -220,7 +221,7 @@ public void run() {
consumer.subscribe(Collections.singletonList(topic), this);

while (!isFinished()) {
ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
Map<TopicPartition, OffsetAndMetadata> offsets = onRecordsReceived(records);

if (!useAutoCommit) {
Expand Down
3 changes: 2 additions & 1 deletion tools/src/main/java/org/apache/kafka/trogdor/workload/ConsumeBenchWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@

import org.apache.kafka.trogdor.task.TaskWorker;

import java.time.Duration;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
Expand Down Expand Up @@ -135,7 +136,7 @@ public Void call() throws Exception {
long startBatchMs = startTimeMs;
try {
while (messagesConsumed < spec.maxMessages()) {
ConsumerRecords<byte[], byte[]> records = consumer.poll(50);
ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(50));
if (records.isEmpty()) {
continue;
}
Expand Down
3 changes: 2 additions & 1 deletion tools/src/main/java/org/apache/kafka/trogdor/workload/RoundTripWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
Expand Down Expand Up @@ -337,7 +338,7 @@ public void run() {
while (true) {
try {
pollInvoked++;
ConsumerRecords<byte[], byte[]> records = consumer.poll(50);
ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(50));
for (Iterator<ConsumerRecord<byte[], byte[]>> iter = records.iterator(); iter.hasNext(); ) {
ConsumerRecord<byte[], byte[]> record = iter.next();
int messageIndex = ByteBuffer.wrap(record.key()).order(ByteOrder.LITTLE_ENDIAN).getInt();
Expand Down