cameronlee314 commented on a change in pull request #951: SAMZA-2127: Upgrade
to Kafka 2.0
URL: https://github.com/apache/samza/pull/951#discussion_r267095405
##########
File path:
samza-test/src/test/scala/org/apache/samza/test/integration/StreamTaskTestUtil.scala
##########
@@ -257,23 +250,27 @@ class StreamTaskTestUtil {
props.put("group.id", group)
props.put("auto.offset.reset", "smallest")
- val consumerConfig = new ConsumerConfig(props)
- val consumerConnector = Consumer.create(consumerConfig)
- val stream = consumerConnector.createMessageStreams(Map(topic ->
1))(topic).head.iterator
- var message: MessageAndMetadata[Array[Byte], Array[Byte]] = null
+ val consumerConnector = new KafkaConsumer(props)
+ consumerConnector.subscribe(Set(topic).asJava)
+
+ var stream = consumerConnector.poll(Duration.ofMillis(100)).iterator()
+ var message: ConsumerRecord[Nothing, Nothing] = null
var messages = ArrayBuffer[String]()
while (message == null || message.offset < maxOffsetInclusive) {
- message = stream.next
- if (message.message == null) {
- messages += null
+ if (stream.hasNext) {
+ message = stream.next
+ if (message.value() == null) {
+ messages += null
+ } else {
+ messages += new String(message.value, "UTF-8")
+ }
+ System.err.println("StreamTaskTestUtil.readAll(): offset=%s,
message=%s" format (message.offset, messages.last))
Review comment:
I know this was already here, but it doesn't seem like an error. Could it
just be removed, or at least changed to System.out?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
With regards,
Apache Git Services
