Skip to content

MINOR: Fix handling of dummy record in EndToEndLatency tool #5950

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

Closed
apovzner wants to merge 3 commits into from
Closed

MINOR: Fix handling of dummy record in EndToEndLatency tool #5950

Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4ae562e
MINOR: Fix handling of dummy record in EndToEndLatency tool
apovzner Nov 27, 2018
5a2f31f
Using AdminClient to create topic instead of relying on auto-creation
apovzner Nov 28, 2018
310e9d0
Loading props from user to AdminClient config
apovzner Nov 28, 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
31 changes: 27 additions & 4 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 @@ -19,9 +19,11 @@ package kafka.tools

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

import kafka.utils.Exit
import org.apache.kafka.clients.admin.NewTopic
import org.apache.kafka.clients.{admin, CommonClientConfigs}
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import org.apache.kafka.clients.producer._
import org.apache.kafka.common.TopicPartition
Expand All @@ -44,6 +46,8 @@ import scala.util.Random

object EndToEndLatency {
private val timeout: Long = 60000
private val defaultReplicationFactor: Short = 1
private val defaultNumPartitions: Int = 1

def main(args: Array[String]) {
if (args.length != 5 && args.length != 6) {
Expand Down Expand Up @@ -82,15 +86,22 @@ object EndToEndLatency {
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer")
val producer = new KafkaProducer[Array[Byte], Array[Byte]](producerProps)

// sends a dummy message to create the topic if it doesn't exist
producer.send(new ProducerRecord[Array[Byte], Array[Byte]](topic, Array[Byte]())).get()

def finalise() {
consumer.commitSync()
producer.close()
consumer.close()
}

// create topic if it does not exist
if (!consumer.listTopics().containsKey(topic)) {
try {
createTopic(topic, brokerList)
} catch {
case t: Throwable =>
finalise()
throw new RuntimeException(s"Failed to create topic $topic", t)
}
}

val topicPartitions = consumer.partitionsFor(topic).asScala
.map(p => new TopicPartition(p.topic(), p.partition())).asJava
Expand Down Expand Up @@ -153,4 +164,16 @@ object EndToEndLatency {
def randomBytesOfLen(random: Random, len: Int): Array[Byte] = {
Array.fill(len)((random.nextInt(26) + 65).toByte)
}

def createTopic(topic: String, brokerList: String): Unit = {
println("Topic \"%s\" does not exist. Will create topic with %d partition(s) and replication factor = %d"
.format(topic, defaultNumPartitions, defaultReplicationFactor))

val props = new Properties()

@ewencp ewencp Nov 28, 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 think you want to loadProps here like the producer and consumer do. Otherwise this wouldn't work with, e.g. security. Not sure if we're actually leveraging extra props anywhere, but I think we probably want to be consistent.

@apovzner apovzner Nov 28, 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.

Ah good point about security and consistency. Will fix.

props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerList)
val adminClient = admin.AdminClient.create(props)
val newTopic = new NewTopic(topic, defaultNumPartitions, defaultReplicationFactor)
try adminClient.createTopics(Collections.singleton(newTopic)).all().get()
finally Utils.closeQuietly(adminClient, "AdminClient")
}
}