You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Eric Deliot <er...@gmail.com> on 2012/07/24 18:05:56 UTC
compile error "illegal cyclic reference involving class Producer"
with simple example
Hi,
I am just starting with kafka. I've downloaded release 0.7.1 and followed
the quick start instructions to get going without modifying anything.
./sbt update &./sbt package ran fine
I can run the console producer and consumer scripts fine.
However, when I try to create my own producer example I end up with some
"illegal cyclic reference involving class Producer" error at compile time.
As an example, I've copied the file kafka/tools/ProducerShell.scala to a
different directory and renamed it MyProducerShell.scala (removing the
package and renaming the object name).
I copy below the copied file and error stack.
kafka/tools/ProducerShell.scala compiled fine when I ran ./sbt package at
the start so why isn't MyProducerShell.scala compiling? what am I doing
wrong?
Thanks.
Eric
--
ed@edvm-5:~/testkafka$ ls
jopt-simple-3.2.jar kafka-0.7.1.jar MyProducerShell.scala
ed@edvm-5:~/testkafka$ scalac -version
Scala compiler version 2.8.0.final -- Copyright 2002-2010, LAMP/EPFL
ed@edvm-5:~/testkafka$ cat MyProducerShell.scala
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
//package kafka.tools
import java.io._
import joptsimple._
import kafka.producer._
import kafka.utils.Utils
/**
* Interactive shell for producing messages from the command line
*/
object MyProducerShell {
def main(args: Array[String]) {
val parser = new OptionParser
val producerPropsOpt = parser.accepts("props", "REQUIRED: Properties
file with the producer properties.")
.withRequiredArg
.describedAs("properties")
.ofType(classOf[String])
val topicOpt = parser.accepts("topic", "REQUIRED: The topic to produce
to.")
.withRequiredArg
.describedAs("topic")
.ofType(classOf[String])
val options = parser.parse(args : _*)
for(arg <- List(producerPropsOpt, topicOpt)) {
if(!options.has(arg)) {
System.err.println("Missing required argument \"" + arg + "\"")
parser.printHelpOn(System.err)
System.exit(1)
}
}
val propsFile = options.valueOf(producerPropsOpt)
val producerConfig = new ProducerConfig(Utils.loadProps(propsFile))
val topic = options.valueOf(topicOpt)
val producer = new Producer[String, String](producerConfig)
val input = new BufferedReader(new InputStreamReader(System.in))
var done = false
while(!done) {
val line = input.readLine()
if(line == null) {
done = true
} else {
val message = line.trim
producer.send(new ProducerData[String, String](topic, message))
println("Sent: %s (%d bytes)".format(line, message.getBytes.length))
}
}
producer.close()
}
}
ed@edvm-5:~/testkafka$ scalac -cp ".:./*" MyProducerShell.scala
Exception in thread "main" scala.tools.nsc.symtab.Types$TypeError: illegal
cyclic reference involving class Producer
at
scala.tools.nsc.symtab.Types$CompoundType.baseClasses(Types.scala:1271)
at scala.tools.nsc.symtab.Types$PolyType.baseClasses(Types.scala:1978)
at scala.tools.nsc.symtab.Symbols$Symbol.ancestors(Symbols.scala:1193)
at
scala.tools.nsc.symtab.Symbols$Symbol.mixinClasses(Symbols.scala:1188)
at
scala.tools.nsc.transform.ExplicitOuter.transformInfo(ExplicitOuter.scala:143)
at
scala.tools.nsc.transform.ExplicitOuter.transformInfo(ExplicitOuter.scala:155)
at
scala.tools.nsc.transform.InfoTransform$Phase$$anon$1.transform(InfoTransform.scala:35)
at scala.tools.nsc.symtab.Symbols$Symbol.rawInfo(Symbols.scala:810)
at
scala.tools.nsc.symtab.Symbols$Symbol.unsafeTypeParams(Symbols.scala:919)
at
scala.tools.nsc.symtab.Types$TypeRef.typeParamsDirect(Types.scala:1686)
at scala.tools.nsc.symtab.Types$TypeRef.isHigherKinded(Types.scala:1700)
at scala.tools.nsc.symtab.Types$TypeRef.normalize0(Types.scala:1739)
at scala.tools.nsc.symtab.Types$TypeRef.normalize(Types.scala:1761)
at
scala.tools.nsc.symtab.Definitions$definitions$.isFunctionType(Definitions.scala:341)
at scala.tools.nsc.symtab.Types$TypeRef.safeToString(Types.scala:1828)
at scala.reflect.generic.Types$AbsType.toString(Types.scala:27)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:348)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printOpt(TreePrinters.scala:113)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:172)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printColumn(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:239)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printOpt(TreePrinters.scala:113)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:183)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:44)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printColumn(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:234)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:156)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:44)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printColumn(TreePrinters.scala:51)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:160)
at
scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
at scala.reflect.generic.Trees$Tree.toString(Trees.scala:135)
at java.lang.String.valueOf(String.java:2826)
at
scala.collection.mutable.StringBuilder.append(StringBuilder.scala:178)
at
scala.tools.nsc.transform.Erasure$Eraser.liftedTree1$1(Erasure.scala:656)
at scala.tools.nsc.transform.Erasure$Eraser.typed1(Erasure.scala:648)
at scala.tools.nsc.typechecker.Typers$Typer.typed(Typers.scala:4078)
at scala.tools.nsc.typechecker.Typers$Typer.typed(Typers.scala:4126)
at
scala.tools.nsc.transform.Erasure$ErasureTransformer$$anonfun$transform$1.apply(Erasure.scala:1082)
at
scala.tools.nsc.transform.Erasure$ErasureTransformer$$anonfun$transform$1.apply(Erasure.scala:1079)
at scala.tools.nsc.symtab.SymbolTable.atPhase(SymbolTable.scala:103)
at
scala.tools.nsc.transform.Erasure$ErasureTransformer.transform(Erasure.scala:1079)
at scala.tools.nsc.ast.Trees$Transformer.transformUnit(Trees.scala:857)
at scala.tools.nsc.transform.Transform$Phase.apply(Transform.scala:30)
at
scala.tools.nsc.Global$GlobalPhase$$anonfun$applyPhase$1.apply(Global.scala:276)
at
scala.tools.nsc.Global$GlobalPhase$$anonfun$applyPhase$1.apply(Global.scala:276)
at scala.tools.nsc.reporters.Reporter.withSource(Reporter.scala:48)
at scala.tools.nsc.Global$GlobalPhase.applyPhase(Global.scala:276)
at
scala.tools.nsc.Global$GlobalPhase$$anonfun$run$1.apply(Global.scala:252)
at
scala.tools.nsc.Global$GlobalPhase$$anonfun$run$1.apply(Global.scala:252)
at scala.collection.Iterator$class.foreach(Iterator.scala:631)
at
scala.collection.mutable.ListBuffer$$anon$1.foreach(ListBuffer.scala:304)
at scala.tools.nsc.Global$GlobalPhase.run(Global.scala:252)
at scala.tools.nsc.Global$Run.compileSources(Global.scala:733)
at scala.tools.nsc.Global$Run.compile(Global.scala:813)
at scala.tools.nsc.Main$.process(Main.scala:108)
at scala.tools.nsc.Main$.main(Main.scala:122)
at scala.tools.nsc.Main.main(Main.scala)
ed@edvm-5:~/testkafka$
Re: compile error "illegal cyclic reference involving class Producer" with simple example
Posted by Sam William <sa...@stumbleupon.com>.
You probably might have solved this or you do not need this anymore.. anyway , I got this exact same error , when I built my projects with Kafka(I have this built with scala-2.9.2) as a dependancy and using scala-2.9.2. Adding log4 as a dependancy seems to have solved it.
Thanks,
Sam
On Jul 24, 2012, at 9:05 AM, Eric Deliot <er...@gmail.com> wrote:
> Hi,
>
> I am just starting with kafka. I've downloaded release 0.7.1 and followed
> the quick start instructions to get going without modifying anything.
> ./sbt update &./sbt package ran fine
> I can run the console producer and consumer scripts fine.
> However, when I try to create my own producer example I end up with some
> "illegal cyclic reference involving class Producer" error at compile time.
> As an example, I've copied the file kafka/tools/ProducerShell.scala to a
> different directory and renamed it MyProducerShell.scala (removing the
> package and renaming the object name).
> I copy below the copied file and error stack.
> kafka/tools/ProducerShell.scala compiled fine when I ran ./sbt package at
> the start so why isn't MyProducerShell.scala compiling? what am I doing
> wrong?
> Thanks.
>
> Eric
>
> --
> ed@edvm-5:~/testkafka$ ls
> jopt-simple-3.2.jar kafka-0.7.1.jar MyProducerShell.scala
>
> ed@edvm-5:~/testkafka$ scalac -version
> Scala compiler version 2.8.0.final -- Copyright 2002-2010, LAMP/EPFL
>
> ed@edvm-5:~/testkafka$ cat MyProducerShell.scala
> /**
> * Licensed to the Apache Software Foundation (ASF) under one or more
> * contributor license agreements. See the NOTICE file distributed with
> * this work for additional information regarding copyright ownership.
> * The ASF licenses this file to You under the Apache License, Version 2.0
> * (the "License"); you may not use this file except in compliance with
> * the License. You may obtain a copy of the License at
> *
> * http://www.apache.org/licenses/LICENSE-2.0
> *
> * Unless required by applicable law or agreed to in writing, software
> * distributed under the License is distributed on an "AS IS" BASIS,
> * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> * See the License for the specific language governing permissions and
> * limitations under the License.
> */
>
> //package kafka.tools
>
> import java.io._
> import joptsimple._
> import kafka.producer._
> import kafka.utils.Utils
>
> /**
> * Interactive shell for producing messages from the command line
> */
> object MyProducerShell {
>
> def main(args: Array[String]) {
>
> val parser = new OptionParser
> val producerPropsOpt = parser.accepts("props", "REQUIRED: Properties
> file with the producer properties.")
> .withRequiredArg
> .describedAs("properties")
> .ofType(classOf[String])
> val topicOpt = parser.accepts("topic", "REQUIRED: The topic to produce
> to.")
> .withRequiredArg
> .describedAs("topic")
> .ofType(classOf[String])
>
> val options = parser.parse(args : _*)
>
> for(arg <- List(producerPropsOpt, topicOpt)) {
> if(!options.has(arg)) {
> System.err.println("Missing required argument \"" + arg + "\"")
> parser.printHelpOn(System.err)
> System.exit(1)
> }
> }
>
> val propsFile = options.valueOf(producerPropsOpt)
> val producerConfig = new ProducerConfig(Utils.loadProps(propsFile))
> val topic = options.valueOf(topicOpt)
> val producer = new Producer[String, String](producerConfig)
>
> val input = new BufferedReader(new InputStreamReader(System.in))
> var done = false
> while(!done) {
> val line = input.readLine()
> if(line == null) {
> done = true
> } else {
> val message = line.trim
> producer.send(new ProducerData[String, String](topic, message))
> println("Sent: %s (%d bytes)".format(line, message.getBytes.length))
> }
> }
> producer.close()
> }
> }
>
> ed@edvm-5:~/testkafka$ scalac -cp ".:./*" MyProducerShell.scala
> Exception in thread "main" scala.tools.nsc.symtab.Types$TypeError: illegal
> cyclic reference involving class Producer
> at
> scala.tools.nsc.symtab.Types$CompoundType.baseClasses(Types.scala:1271)
> at scala.tools.nsc.symtab.Types$PolyType.baseClasses(Types.scala:1978)
> at scala.tools.nsc.symtab.Symbols$Symbol.ancestors(Symbols.scala:1193)
> at
> scala.tools.nsc.symtab.Symbols$Symbol.mixinClasses(Symbols.scala:1188)
> at
> scala.tools.nsc.transform.ExplicitOuter.transformInfo(ExplicitOuter.scala:143)
> at
> scala.tools.nsc.transform.ExplicitOuter.transformInfo(ExplicitOuter.scala:155)
> at
> scala.tools.nsc.transform.InfoTransform$Phase$$anon$1.transform(InfoTransform.scala:35)
> at scala.tools.nsc.symtab.Symbols$Symbol.rawInfo(Symbols.scala:810)
> at
> scala.tools.nsc.symtab.Symbols$Symbol.unsafeTypeParams(Symbols.scala:919)
> at
> scala.tools.nsc.symtab.Types$TypeRef.typeParamsDirect(Types.scala:1686)
> at scala.tools.nsc.symtab.Types$TypeRef.isHigherKinded(Types.scala:1700)
> at scala.tools.nsc.symtab.Types$TypeRef.normalize0(Types.scala:1739)
> at scala.tools.nsc.symtab.Types$TypeRef.normalize(Types.scala:1761)
> at
> scala.tools.nsc.symtab.Definitions$definitions$.isFunctionType(Definitions.scala:341)
> at scala.tools.nsc.symtab.Types$TypeRef.safeToString(Types.scala:1828)
> at scala.reflect.generic.Types$AbsType.toString(Types.scala:27)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:348)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printOpt(TreePrinters.scala:113)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:172)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printColumn(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:239)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printOpt(TreePrinters.scala:113)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:183)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:44)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:45)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printColumn(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:234)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:156)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter$$anonfun$printColumn$2.apply(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printSeq(TreePrinters.scala:44)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printColumn(TreePrinters.scala:51)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.printRaw(TreePrinters.scala:160)
> at
> scala.tools.nsc.ast.TreePrinters$TreePrinter.print(TreePrinters.scala:392)
> at scala.reflect.generic.Trees$Tree.toString(Trees.scala:135)
> at java.lang.String.valueOf(String.java:2826)
> at
> scala.collection.mutable.StringBuilder.append(StringBuilder.scala:178)
> at
> scala.tools.nsc.transform.Erasure$Eraser.liftedTree1$1(Erasure.scala:656)
> at scala.tools.nsc.transform.Erasure$Eraser.typed1(Erasure.scala:648)
> at scala.tools.nsc.typechecker.Typers$Typer.typed(Typers.scala:4078)
> at scala.tools.nsc.typechecker.Typers$Typer.typed(Typers.scala:4126)
> at
> scala.tools.nsc.transform.Erasure$ErasureTransformer$$anonfun$transform$1.apply(Erasure.scala:1082)
> at
> scala.tools.nsc.transform.Erasure$ErasureTransformer$$anonfun$transform$1.apply(Erasure.scala:1079)
> at scala.tools.nsc.symtab.SymbolTable.atPhase(SymbolTable.scala:103)
> at
> scala.tools.nsc.transform.Erasure$ErasureTransformer.transform(Erasure.scala:1079)
> at scala.tools.nsc.ast.Trees$Transformer.transformUnit(Trees.scala:857)
> at scala.tools.nsc.transform.Transform$Phase.apply(Transform.scala:30)
> at
> scala.tools.nsc.Global$GlobalPhase$$anonfun$applyPhase$1.apply(Global.scala:276)
> at
> scala.tools.nsc.Global$GlobalPhase$$anonfun$applyPhase$1.apply(Global.scala:276)
> at scala.tools.nsc.reporters.Reporter.withSource(Reporter.scala:48)
> at scala.tools.nsc.Global$GlobalPhase.applyPhase(Global.scala:276)
> at
> scala.tools.nsc.Global$GlobalPhase$$anonfun$run$1.apply(Global.scala:252)
> at
> scala.tools.nsc.Global$GlobalPhase$$anonfun$run$1.apply(Global.scala:252)
> at scala.collection.Iterator$class.foreach(Iterator.scala:631)
> at
> scala.collection.mutable.ListBuffer$$anon$1.foreach(ListBuffer.scala:304)
> at scala.tools.nsc.Global$GlobalPhase.run(Global.scala:252)
> at scala.tools.nsc.Global$Run.compileSources(Global.scala:733)
> at scala.tools.nsc.Global$Run.compile(Global.scala:813)
> at scala.tools.nsc.Main$.process(Main.scala:108)
> at scala.tools.nsc.Main$.main(Main.scala:122)
> at scala.tools.nsc.Main.main(Main.scala)
> ed@edvm-5:~/testkafka$