[ https://issues.apache.org/jira/browse/KAFKA-5327?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16377427#comment-16377427 ]
ASF GitHub Bot commented on KAFKA-5327: --------------------------------------- hachikuji closed pull request #4546: KAFKA-5327: Console Consumer should only poll for up to max messages URL: https://github.com/apache/kafka/pull/4546 This is a PR merged from a forked repository. As GitHub hides the original diff on merge, it is displayed below for the sake of provenance: As this is a foreign pull request (from a fork), the diff is supplied below (as it won't show otherwise due to GitHub magic): diff --git a/core/src/main/scala/kafka/consumer/BaseConsumer.scala b/core/src/main/scala/kafka/consumer/BaseConsumer.scala index 04ac2d9f1a5..9e49fe4dc8b 100644 --- a/core/src/main/scala/kafka/consumer/BaseConsumer.scala +++ b/core/src/main/scala/kafka/consumer/BaseConsumer.scala @@ -23,6 +23,7 @@ import java.util.regex.Pattern import kafka.api.OffsetRequest import kafka.common.StreamEndException import kafka.message.Message +import org.apache.kafka.clients.consumer.Consumer import org.apache.kafka.common.record.TimestampType import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.header.Headers @@ -55,10 +56,8 @@ case class BaseConsumerRecord(topic: String, @deprecated("This class has been deprecated and will be removed in a future release. " + "Please use org.apache.kafka.clients.consumer.KafkaConsumer instead.", "0.11.0.0") -class NewShinyConsumer(topic: Option[String], partitionId: Option[Int], offset: Option[Long], whitelist: Option[String], consumerProps: Properties, val timeoutMs: Long = Long.MaxValue) extends BaseConsumer { - import org.apache.kafka.clients.consumer.KafkaConsumer - - val consumer = new KafkaConsumer[Array[Byte], Array[Byte]](consumerProps) +class NewShinyConsumer(topic: Option[String], partitionId: Option[Int], offset: Option[Long], whitelist: Option[String], + consumer: Consumer[Array[Byte], Array[Byte]], val timeoutMs: Long = Long.MaxValue) extends BaseConsumer { consumerInit() var recordIter = consumer.poll(0).iterator @@ -91,6 +90,17 @@ class NewShinyConsumer(topic: Option[String], partitionId: Option[Int], offset: } } + def resetUnconsumedOffsets() { + val smallestUnconsumedOffsets = collection.mutable.Map[TopicPartition, Long]() + while (recordIter.hasNext) { + val record = recordIter.next() + val tp = new TopicPartition(record.topic, record.partition) + // avoid auto-committing offsets which haven't been consumed + smallestUnconsumedOffsets.getOrElseUpdate(tp, record.offset) + } + smallestUnconsumedOffsets.foreach { case (tp, offset) => consumer.seek(tp, offset) } + } + override def receive(): BaseConsumerRecord = { if (!recordIter.hasNext) { recordIter = consumer.poll(timeoutMs).iterator @@ -114,6 +124,7 @@ class NewShinyConsumer(topic: Option[String], partitionId: Option[Int], offset: } override def cleanup() { + resetUnconsumedOffsets() this.consumer.close() } diff --git a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala index 7d2d3710157..24fa583f1bb 100755 --- a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala +++ b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala @@ -31,10 +31,10 @@ import kafka.message._ import kafka.metrics.KafkaMetricsReporter import kafka.utils._ import kafka.utils.Implicits._ -import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord} +import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, KafkaConsumer} import org.apache.kafka.common.errors.{AuthenticationException, WakeupException} import org.apache.kafka.common.record.TimestampType -import org.apache.kafka.common.serialization.Deserializer +import org.apache.kafka.common.serialization.{ByteArrayDeserializer, Deserializer} import org.apache.kafka.common.utils.Utils import scala.collection.JavaConverters._ @@ -72,10 +72,11 @@ object ConsoleConsumer extends Logging { new OldConsumer(conf.filterSpec, props) } else { val timeoutMs = if (conf.timeoutMs >= 0) conf.timeoutMs else Long.MaxValue + val consumer = new KafkaConsumer(getNewConsumerProps(conf), new ByteArrayDeserializer, new ByteArrayDeserializer) if (conf.partitionArg.isDefined) - new NewShinyConsumer(Option(conf.topicArg), conf.partitionArg, Option(conf.offsetArg), None, getNewConsumerProps(conf), timeoutMs) + new NewShinyConsumer(Option(conf.topicArg), conf.partitionArg, Option(conf.offsetArg), None, consumer, timeoutMs) else - new NewShinyConsumer(Option(conf.topicArg), None, None, Option(conf.whitelistArg), getNewConsumerProps(conf), timeoutMs) + new NewShinyConsumer(Option(conf.topicArg), None, None, Option(conf.whitelistArg), consumer, timeoutMs) } addShutdownHook(consumer, conf) @@ -202,15 +203,12 @@ object ConsoleConsumer extends Logging { } } - def getNewConsumerProps(config: ConsumerConfig): Properties = { + private[tools] def getNewConsumerProps(config: ConsumerConfig): Properties = { val props = new Properties - props ++= config.consumerProps props ++= config.extraConsumerProps setAutoOffsetResetValue(config, props) props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.bootstrapServer) - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer") - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer") props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, config.isolationLevel) props } diff --git a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala index 364f6b3b93e..9ae8b966ac0 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala @@ -17,18 +17,65 @@ package kafka.tools -import java.io.{PrintStream, FileOutputStream} +import java.io.{FileOutputStream, PrintStream} import kafka.common.MessageFormatter -import kafka.consumer.{BaseConsumer, BaseConsumerRecord} +import kafka.consumer.{BaseConsumer, BaseConsumerRecord, NewShinyConsumer} import kafka.utils.{Exit, TestUtils} +import org.apache.kafka.clients.consumer.{ConsumerRecord, MockConsumer, OffsetResetStrategy} +import org.apache.kafka.common.TopicPartition import org.apache.kafka.clients.consumer.ConsumerConfig import org.easymock.EasyMock import org.junit.Assert._ -import org.junit.Test +import org.junit.{Before, Test} + +import scala.collection.JavaConverters._ class ConsoleConsumerTest { + @Before + def setup(): Unit = { + ConsoleConsumer.messageCount = 0 + } + + @Test + def shouldResetUnConsumedOffsetsBeforeExitForNewConsumer() { + val topic = "test" + val maxMessages: Int = 123 + val totalMessages: Int = 700 + val startOffset: java.lang.Long = 0L + + val mockConsumer = new MockConsumer[Array[Byte], Array[Byte]](OffsetResetStrategy.EARLIEST) + val tp1 = new TopicPartition(topic, 0) + val tp2 = new TopicPartition(topic, 1) + + val consumer = new NewShinyConsumer(Some(topic), None, None, None, mockConsumer) + + mockConsumer.rebalance(List(tp1, tp2).asJava) + mockConsumer.updateBeginningOffsets(Map(tp1 -> startOffset, tp2 -> startOffset).asJava) + + 0 until totalMessages foreach { i => + // add all records, each partition should have half of `totalMessages` + mockConsumer.addRecord(new ConsumerRecord[Array[Byte], Array[Byte]](topic, i % 2, i / 2, "key".getBytes, "value".getBytes)) + } + + // Mocks + val formatter = EasyMock.createNiceMock(classOf[MessageFormatter]) + + // Expectations + EasyMock.expect(formatter.writeTo(EasyMock.anyObject(), EasyMock.anyObject())).times(maxMessages) + EasyMock.replay(formatter) + + // Test + ConsoleConsumer.process(maxMessages, formatter, consumer, System.out, skipMessageOnError = false) + assertEquals(totalMessages, mockConsumer.position(tp1) + mockConsumer.position(tp2)) + + consumer.resetUnconsumedOffsets() + assertEquals(maxMessages, mockConsumer.position(tp1) + mockConsumer.position(tp2)) + + EasyMock.verify(formatter) + } + @Test def shouldLimitReadsToMaxMessageLimit() { //Mocks ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Console Consumer should only poll for up to max messages > -------------------------------------------------------- > > Key: KAFKA-5327 > URL: https://issues.apache.org/jira/browse/KAFKA-5327 > Project: Kafka > Issue Type: Improvement > Components: tools > Reporter: Dustin Cote > Assignee: huxihx > Priority: Minor > Fix For: 1.2.0 > > > The ConsoleConsumer has a --max-messages flag that can be used to limit the > number of messages consumed. However, the number of records actually consumed > is governed by max.poll.records. This means you see one message on the > console, but your offset has moved forward a default of 500, which is kind of > counterintuitive. It would be good to only commit offsets for messages we > have printed to the console. -- This message was sent by Atlassian JIRA (v7.6.3#76005)