[jira] [Commented] (KAFKA-1494) Failed to send messages after 3 tries.

[Josh Rosen (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1494?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154518#comment-14154518
 ] 

Josh Rosen commented on KAFKA-1494:
---

I'm able to reproduce a similar problem using Kafka 0.8 with a single broker 
running inside of Docker (using boot2docker) on OS X.  In this setup, I run a 
single-node ZooKeeper process in its own Docker container and configure the 
Kafka broker to use it.  The broker is configured to bind to the Docker 
container's IP on the VirtualBox network.  Using {{sudo route \-n add 
172.17.0.0/16 `boot2docker ip`}}, I've configured my Mac so that programs 
running outside of Docker can access network services running inside of 
containers using the containers' own IP addresses (so we use the same IPs for 
host-container and container-container communication).  I have some Scala code 
which uses Kafka's ZkUtils to retrieve information on the Kafka cluster state 
and kafka.producer.Producer to push data into Kafka.

When I attempt to run this, I see errors very similar to the ones reported in 
this ticket:

{code}
14/10/01 00:22:54 INFO client.ClientUtils$: Fetching metadata from broker 
id:0,host:172.17.0.187,port:12345 with correlation id 11 for 1 topic(s) 
Set(test-topic)
14/10/01 00:22:54 INFO producer.SyncProducer: Connected to 172.17.0.187:12345 
for producing
14/10/01 00:22:54 INFO producer.SyncProducer: Disconnecting from 
172.17.0.187:12345
14/10/01 00:22:54 WARN producer.BrokerPartitionInfo: Error while fetching 
metadata [{TopicMetadata for topic test-topic ->
No partition metadata for topic test-topic due to 
kafka.common.LeaderNotAvailableException}] for topic [test-topic]: class 
kafka.common.LeaderNotAvailableException
{code}

I've found messages on the mailing list that describe this 
LeaderNotAvailableException; however, I don't think that any of the root causes 
reported there apply in my case:


- My producer can connect to the broker (I know this because I set `client.id` 
in my producer configuration and see matching entries in the broker's logs).
- My broker can connect to ZooKeeper (I manually inspected ZooKeeper's contents 
using zookeeper-shell).
- I didn't attempt to delete any topics.
- ZooKeeper starts with a completely fresh state; leftover state from earlier 
attempts / experiments isn't an issue here.
- I don't think there are any race-conditions in my client code; I added long 
(20-30 second) Thread.sleep() calls in my program so that we wait for leader 
election / metadata propagation to occur after creating the new topic.


I've noticed that ZkUtils.getReplicasForPartition lists the expected replica 
for the partition, although ZkUtils.getLeaderForPartition returns None, even 
after multiple attempts / long delays.

This test is easy for me to re-run; the entire setup / teardown of the 
containers takes less than a minute.  Therefore, it's very easy for me to 
experiment with different configurations / settings.  I can provide any logs, 
configuration files, etc. that would be helpful when debugging this; just let 
me know what you need.

> Failed to send messages after 3 tries.
> --
>
> Key: KAFKA-1494
> URL: https://issues.apache.org/jira/browse/KAFKA-1494
> Project: Kafka
>  Issue Type: Bug
>  Components: controller, core
>Affects Versions: 0.8.1.1
> Environment: Mac OS 
>Reporter: darion yaphets
>Assignee: Neha Narkhede
>
> I use default server & zookeeper config to start-up zookeeper server and 
> kafka broker on my machine to test custom message which based on proto buffer 
> . I write a client to send protobuf-message to kafka broker and source code 
> as following :
> Properties properties = new Properties();
>   properties.put("serializer.class", 
> "java_example.ProtoBufMessage");
>   properties.put("metadata.broker.list", "localhost:9092");
>   ProducerConfig config = new ProducerConfig(properties);
>   testBuf buffer = testBuf.newBuilder().setID(0)
>   .setUrl("darion.yaphet.org").build();
>   Producer producer = new Producer testBuf>(
>   config);
>   producer.send(new KeyedMessage("protobuffer", 
> buffer));
> client debug log report a exception:
> [FileSystemMoniter] INFO [main] kafka.utils.Logging$class.info(68) | 
> Disconnecting from localhost:9092
> [FileSystemMoniter] DEBUG [main] kafka.utils.Logging$class.debug(52) | 
> Successfully fetched metadata for 1 topic(s) Set(protobuffer)
> [FileSystemMoniter] WARN [main] kafka.utils.Logging$class.warn(83) | Error 
> while fetching metadata [{TopicMetadata for topic protobuffer -> 
> No partition metadata for topic protobuffer due to 
> kafka.common.LeaderNotAvailableException}] for topic [protobuffer]: cl

[jira] [Commented] (KAFKA-1494) Failed to send messages after 3 tries.

[Andras Hatvani (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1494?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154569#comment-14154569
 ] 

Andras Hatvani commented on KAFKA-1494:
---

Josh, I've got a very similar Docker-based setup and I was able to successfully 
produce and consume messages by adjusting the variables I described in my 
previous comment in this JIRA issue. I suggest that you send this problem 
description to the user mailing list where we can continue the discussion 
without polluting this issue.

> Failed to send messages after 3 tries.
> --
>
> Key: KAFKA-1494
> URL: https://issues.apache.org/jira/browse/KAFKA-1494
> Project: Kafka
>  Issue Type: Bug
>  Components: controller, core
>Affects Versions: 0.8.1.1
> Environment: Mac OS 
>Reporter: darion yaphets
>Assignee: Neha Narkhede
>
> I use default server & zookeeper config to start-up zookeeper server and 
> kafka broker on my machine to test custom message which based on proto buffer 
> . I write a client to send protobuf-message to kafka broker and source code 
> as following :
> Properties properties = new Properties();
>   properties.put("serializer.class", 
> "java_example.ProtoBufMessage");
>   properties.put("metadata.broker.list", "localhost:9092");
>   ProducerConfig config = new ProducerConfig(properties);
>   testBuf buffer = testBuf.newBuilder().setID(0)
>   .setUrl("darion.yaphet.org").build();
>   Producer producer = new Producer testBuf>(
>   config);
>   producer.send(new KeyedMessage("protobuffer", 
> buffer));
> client debug log report a exception:
> [FileSystemMoniter] INFO [main] kafka.utils.Logging$class.info(68) | 
> Disconnecting from localhost:9092
> [FileSystemMoniter] DEBUG [main] kafka.utils.Logging$class.debug(52) | 
> Successfully fetched metadata for 1 topic(s) Set(protobuffer)
> [FileSystemMoniter] WARN [main] kafka.utils.Logging$class.warn(83) | Error 
> while fetching metadata [{TopicMetadata for topic protobuffer -> 
> No partition metadata for topic protobuffer due to 
> kafka.common.LeaderNotAvailableException}] for topic [protobuffer]: class 
> kafka.common.LeaderNotAvailableException 
> [FileSystemMoniter] ERROR [main] kafka.utils.Logging$class.error(97) | Failed 
> to send requests for topics protobuffer with correlation ids in [0,8]
> Exception in thread "main" kafka.common.FailedToSendMessageException: Failed 
> to send messages after 3 tries.
>   at 
> kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
>   at kafka.producer.Producer.send(Producer.scala:76)
>   at kafka.javaapi.producer.Producer.send(Producer.scala:33)
>   at java_example.ProducerExamples.main(ProducerExamples.java:26)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-1661) Move MockConsumer and MockProducer from src/main to src/test

[Andras Hatvani (JIRA)]

Andras Hatvani created KAFKA-1661:
-

 Summary: Move MockConsumer and MockProducer from src/main to 
src/test
 Key: KAFKA-1661
 URL: https://issues.apache.org/jira/browse/KAFKA-1661
 Project: Kafka
  Issue Type: Task
  Components: clients, consumer, producer 
Affects Versions: 0.8.1.1
 Environment: N/A
Reporter: Andras Hatvani
Assignee: Neha Narkhede
 Fix For: 0.8.2


The MockConsumer and MockProducer are currently in src/main although they 
belong in src/test.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-1494) Failed to send messages after 3 tries.

[Joe Stein (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1494?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Joe Stein resolved KAFKA-1494.
--
Resolution: Won't Fix

Please make sure your advertised host name is set 
https://kafka.apache.org/documentation.html#brokerconfigs and should use the 
mailing list https://kafka.apache.org/contact.html fro these type of issues 
(the mailing list just had a docker issues conversation) they are not 
necessarily defects but more likely configuration issues that the community can 
assist you with.

> Failed to send messages after 3 tries.
> --
>
> Key: KAFKA-1494
> URL: https://issues.apache.org/jira/browse/KAFKA-1494
> Project: Kafka
>  Issue Type: Bug
>  Components: controller, core
>Affects Versions: 0.8.1.1
> Environment: Mac OS 
>Reporter: darion yaphets
>Assignee: Neha Narkhede
>
> I use default server & zookeeper config to start-up zookeeper server and 
> kafka broker on my machine to test custom message which based on proto buffer 
> . I write a client to send protobuf-message to kafka broker and source code 
> as following :
> Properties properties = new Properties();
>   properties.put("serializer.class", 
> "java_example.ProtoBufMessage");
>   properties.put("metadata.broker.list", "localhost:9092");
>   ProducerConfig config = new ProducerConfig(properties);
>   testBuf buffer = testBuf.newBuilder().setID(0)
>   .setUrl("darion.yaphet.org").build();
>   Producer producer = new Producer testBuf>(
>   config);
>   producer.send(new KeyedMessage("protobuffer", 
> buffer));
> client debug log report a exception:
> [FileSystemMoniter] INFO [main] kafka.utils.Logging$class.info(68) | 
> Disconnecting from localhost:9092
> [FileSystemMoniter] DEBUG [main] kafka.utils.Logging$class.debug(52) | 
> Successfully fetched metadata for 1 topic(s) Set(protobuffer)
> [FileSystemMoniter] WARN [main] kafka.utils.Logging$class.warn(83) | Error 
> while fetching metadata [{TopicMetadata for topic protobuffer -> 
> No partition metadata for topic protobuffer due to 
> kafka.common.LeaderNotAvailableException}] for topic [protobuffer]: class 
> kafka.common.LeaderNotAvailableException 
> [FileSystemMoniter] ERROR [main] kafka.utils.Logging$class.error(97) | Failed 
> to send requests for topics protobuffer with correlation ids in [0,8]
> Exception in thread "main" kafka.common.FailedToSendMessageException: Failed 
> to send messages after 3 tries.
>   at 
> kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
>   at kafka.producer.Producer.send(Producer.scala:76)
>   at kafka.javaapi.producer.Producer.send(Producer.scala:33)
>   at java_example.ProducerExamples.main(ProducerExamples.java:26)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Comment Edited] (KAFKA-1494) Failed to send messages after 3 tries.

[Joe Stein (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1494?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154732#comment-14154732
 ] 

Joe Stein edited comment on KAFKA-1494 at 10/1/14 12:31 PM:


Please make sure your advertised host name is set 
https://kafka.apache.org/documentation.html#brokerconfigs and you should use 
the mailing list https://kafka.apache.org/contact.html for these type of issues 
(the mailing list just had a docker issues conversation) they are not 
necessarily defects but more likely configuration issues that the community can 
assist you with.  


was (Author: joestein):
Please make sure your advertised host name is set 
https://kafka.apache.org/documentation.html#brokerconfigs and should use the 
mailing list https://kafka.apache.org/contact.html fro these type of issues 
(the mailing list just had a docker issues conversation) they are not 
necessarily defects but more likely configuration issues that the community can 
assist you with.

> Failed to send messages after 3 tries.
> --
>
> Key: KAFKA-1494
> URL: https://issues.apache.org/jira/browse/KAFKA-1494
> Project: Kafka
>  Issue Type: Bug
>  Components: controller, core
>Affects Versions: 0.8.1.1
> Environment: Mac OS 
>Reporter: darion yaphets
>Assignee: Neha Narkhede
>
> I use default server & zookeeper config to start-up zookeeper server and 
> kafka broker on my machine to test custom message which based on proto buffer 
> . I write a client to send protobuf-message to kafka broker and source code 
> as following :
> Properties properties = new Properties();
>   properties.put("serializer.class", 
> "java_example.ProtoBufMessage");
>   properties.put("metadata.broker.list", "localhost:9092");
>   ProducerConfig config = new ProducerConfig(properties);
>   testBuf buffer = testBuf.newBuilder().setID(0)
>   .setUrl("darion.yaphet.org").build();
>   Producer producer = new Producer testBuf>(
>   config);
>   producer.send(new KeyedMessage("protobuffer", 
> buffer));
> client debug log report a exception:
> [FileSystemMoniter] INFO [main] kafka.utils.Logging$class.info(68) | 
> Disconnecting from localhost:9092
> [FileSystemMoniter] DEBUG [main] kafka.utils.Logging$class.debug(52) | 
> Successfully fetched metadata for 1 topic(s) Set(protobuffer)
> [FileSystemMoniter] WARN [main] kafka.utils.Logging$class.warn(83) | Error 
> while fetching metadata [{TopicMetadata for topic protobuffer -> 
> No partition metadata for topic protobuffer due to 
> kafka.common.LeaderNotAvailableException}] for topic [protobuffer]: class 
> kafka.common.LeaderNotAvailableException 
> [FileSystemMoniter] ERROR [main] kafka.utils.Logging$class.error(97) | Failed 
> to send requests for topics protobuffer with correlation ids in [0,8]
> Exception in thread "main" kafka.common.FailedToSendMessageException: Failed 
> to send messages after 3 tries.
>   at 
> kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
>   at kafka.producer.Producer.send(Producer.scala:76)
>   at kafka.javaapi.producer.Producer.send(Producer.scala:33)
>   at java_example.ProducerExamples.main(ProducerExamples.java:26)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-1662) gradle release issue permgen space

[Joe Stein (JIRA)]

Joe Stein created KAFKA-1662:


 Summary: gradle release issue permgen space
 Key: KAFKA-1662
 URL: https://issues.apache.org/jira/browse/KAFKA-1662
 Project: Kafka
  Issue Type: Bug
Reporter: Joe Stein
Priority: Blocker
 Fix For: 0.8.2


Finding issues doing the kafka release with permgen space

ant:scaladoc] java.lang.OutOfMemoryError: PermGen space
:kafka-0.8.2-ALPHA1-src:core:scaladoc FAILED
:releaseTarGz_2_10_1 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Execution failed for task ':core:scaladoc'.
> PermGen space

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 5 mins 55.53 secs

FAILURE: Build failed with an exception.

* What went wrong:
PermGen space





--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: Kafka Trunk Branch build failure

[Joe Stein]

Hi Bhavesh, I reproduced the issue you ran into
https://issues.apache.org/jira/browse/KAFKA-1662 if you want to throw up
your patch that would be great.  Thanks!

/***
 Joe Stein
 Founder, Principal Consultant
 Big Data Open Source Security LLC
 http://www.stealth.ly
 Twitter: @allthingshadoop 
/

On Fri, Sep 26, 2014 at 9:29 PM, Bhavesh Mistry 
wrote:

> Hi Joe,
>
> I run following command.  I am consistently get OO memory for ant forked
> JVM.  I am not able to set JVM argument for
> org.gradle.api.tasks.scala.ScalaDocOptions jvmArgs.  Any Help is greatly
> appreciated.
>
>
> tasks.withType(ScalaDoc) {
> task scaladocJar(type:Jar) {
> classifier = 'scaladoc'
> from '../LICENSE'
> from '../NOTICE'
> from scaladoc
> }
>  configure(scalaDocOptions.additionalParameters) {
> memoryMaximumSize = '1g'
> jvmArgs = ['-XX:MaxPermSize=512m']
> }
>
>
> gradle.properties:
>
> group=org.apache.kafka
> version=0.8.2-SNAPSHOT
> scalaVersion=2.11.2
> task=build
>
> Build Cmd:
>
> ./gradlew uploadArchivesAll
>
>
> Build Error:
>
>
> :kafka:core:scaladoc
> [ant:scaladoc] Element
>
> '/Users/bmistr1/Documents/workspace/kafka-master/kafka/core/build/resources/main'
> does not exist.
> [ant:scaladoc] java.lang.OutOfMemoryError: PermGen space
> [ant:scaladoc] at java.lang.ClassLoader.defineClass1(Native Method)
> [ant:scaladoc] at
> java.lang.ClassLoader.defineClass(ClassLoader.java:800)
> [ant:scaladoc] at
> java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
> [ant:scaladoc] at
> java.net.URLClassLoader.defineClass(URLClassLoader.java:449)
> [ant:scaladoc] at
> java.net.URLClassLoader.access$100(URLClassLoader.java:71)
> [ant:scaladoc] at
> java.net.URLClassLoader$1.run(URLClassLoader.java:361)
> [ant:scaladoc] at
> java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> [ant:scaladoc] at java.security.AccessController.doPrivileged(Native
> Method)
> [ant:scaladoc] at
> java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> [ant:scaladoc] at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> [ant:scaladoc] at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> [ant:scaladoc] at
>
> scala.reflect.internal.ReificationSupport$class.$init$(ReificationSupport.scala:1145)
> [ant:scaladoc] at
> scala.reflect.internal.SymbolTable.(SymbolTable.scala:16)
> [ant:scaladoc] at
> scala.tools.nsc.symtab.SymbolTable.(SymbolTable.scala:9)
> [ant:scaladoc] at scala.tools.nsc.Global.(Global.scala:40)
> [ant:scaladoc] at
> scala.tools.nsc.doc.ScaladocGlobal.(ScaladocGlobal.scala:38)
> [ant:scaladoc] at
> scala.tools.nsc.doc.DocFactory$compiler$.(DocFactory.scala:34)
> [ant:scaladoc] at
> scala.tools.nsc.doc.DocFactory.compiler$lzycompute(DocFactory.scala:34)
> [ant:scaladoc] at
> scala.tools.nsc.doc.DocFactory.compiler(DocFactory.scala:34)
> [ant:scaladoc] at
> scala.tools.nsc.doc.DocFactory.makeUniverse(DocFactory.scala:44)
> [ant:scaladoc] at
> scala.tools.nsc.doc.DocFactory.generate$1(DocFactory.scala:112)
> [ant:scaladoc] at
> scala.tools.nsc.doc.DocFactory.document(DocFactory.scala:124)
> [ant:scaladoc] at scala.tools.ant.Scaladoc.execute(Scaladoc.scala:674)
> [ant:scaladoc] at
> org.apache.tools.ant.UnknownElement.execute(UnknownElement.java:292)
> [ant:scaladoc] at sun.reflect.NativeMethodAccessorImpl.invoke0(Native
> Method)
> [ant:scaladoc] at
>
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> [ant:scaladoc] at
>
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> [ant:scaladoc] at java.lang.reflect.Method.invoke(Method.java:606)
> [ant:scaladoc] at
> org.apache.tools.ant.dispatch.DispatchUtils.execute(DispatchUtils.java:106)
> [ant:scaladoc] at
> groovy.util.AntBuilder.performTask(AntBuilder.java:319)
> [ant:scaladoc] at
> groovy.util.AntBuilder.nodeCompleted(AntBuilder.java:264)
> [ant:scaladoc] at
>
> org.gradle.api.internal.project.ant.BasicAntBuilder.nodeCompleted(BasicAntBuilder.java:71)
>
>
> Thanks,
> Bhavesh
>
>
> On Fri, Sep 26, 2014 at 4:53 PM, Bhavesh Mistry <
> mistry.p.bhav...@gmail.com>
> wrote:
>
> > Hi Joe,
> >
> > Thanks for quick response !
> >
> > Thanks,
> >
> > Bhavesh
> >
> >
> >
> > On Fri, Sep 26, 2014 at 4:19 PM, Joe Stein  wrote:
> >
> >> Hi, yes it was https://issues.apache.org/jira/browse/KAFKA-1490
> >>
> >> https://github.com/apache/kafka/tree/trunk#apache-kafka
> >>
> >> You need to install gradle http://www.gradle.org/installation
> >>
> >> Then bootstrap and download the wrapper
> >>
> >> cd kafka_source_dir
> >> gradle
> >>
> >> Now everything else will work
> >>
> >>
> >> /***
> >>  Joe Stein
> >>  Founder, Principal Consultant
> >>  Big Data Open Source Security LLC
> >>  http://www.s

[jira] [Updated] (KAFKA-1662) gradle release issue permgen space

[Joe Stein (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1662?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Joe Stein updated KAFKA-1662:
-
Labels: newbie  (was: )

> gradle release issue permgen space
> --
>
> Key: KAFKA-1662
> URL: https://issues.apache.org/jira/browse/KAFKA-1662
> Project: Kafka
>  Issue Type: Bug
>Reporter: Joe Stein
>Priority: Blocker
>  Labels: newbie
> Fix For: 0.8.2
>
>
> Finding issues doing the kafka release with permgen space
> ant:scaladoc] java.lang.OutOfMemoryError: PermGen space
> :kafka-0.8.2-ALPHA1-src:core:scaladoc FAILED
> :releaseTarGz_2_10_1 FAILED
> FAILURE: Build failed with an exception.
> * What went wrong:
> Execution failed for task ':core:scaladoc'.
> > PermGen space
> * Try:
> Run with --stacktrace option to get the stack trace. Run with --info or 
> --debug option to get more log output.
> BUILD FAILED
> Total time: 5 mins 55.53 secs
> FAILURE: Build failed with an exception.
> * What went wrong:
> PermGen space



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1660) Ability to call close() with a timeout on the Java Kafka Producer.

[Andrew Stein (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1660?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154809#comment-14154809
 ] 

Andrew Stein commented on KAFKA-1660:
-

Hi [~jkreps],

I don't think that this (the request for {{producer.close(timeout)}}) and 
KAFKA-1659 (the request for {{producer.abort()}}) are the same.
 
If {{producer.close(timeout)}} were to return without actually closing the 
producer in time, the user could then continue with {{producer.abort()}}. 
However, this would not be the only case where {{producer.abort()}} would be 
useful. It could be used if a {{producer.send()}} were to fail or if a 
{{future.get()}} were to fail as well.

In addition if {{producer.close()}} were to return without actually closing the 
producer in time, the user could continue with other logic besides 
{{producer.abort()}}.


> Ability to call close() with a timeout on the Java Kafka Producer. 
> ---
>
> Key: KAFKA-1660
> URL: https://issues.apache.org/jira/browse/KAFKA-1660
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, producer 
>Affects Versions: 0.8.2
>Reporter: Andrew Stein
>Assignee: Jun Rao
> Fix For: 0.8.2
>
>
> I would like the ability to call {{close}} with a timeout on the Java 
> Client's KafkaProducer.
> h6. Workaround
> Currently, it is possible to ensure that {{close}} will return quickly by 
> first doing a {{future.get(timeout)}} on the last future produced on each 
> partition, but this means that the user has to define the partitions up front 
> at the time of {{send}} and track the returned {{future}}'s



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1661) Move MockConsumer and MockProducer from src/main to src/test

[Neha Narkhede (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1661?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neha Narkhede updated KAFKA-1661:
-
Priority: Trivial  (was: Major)

> Move MockConsumer and MockProducer from src/main to src/test
> 
>
> Key: KAFKA-1661
> URL: https://issues.apache.org/jira/browse/KAFKA-1661
> Project: Kafka
>  Issue Type: Task
>  Components: clients, consumer, producer 
>Affects Versions: 0.8.1.1
> Environment: N/A
>Reporter: Andras Hatvani
>Assignee: Neha Narkhede
>Priority: Trivial
>  Labels: test
> Fix For: 0.8.2
>
>
> The MockConsumer and MockProducer are currently in src/main although they 
> belong in src/test.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1661) Move MockConsumer and MockProducer from src/main to src/test

[Neha Narkhede (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1661?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neha Narkhede updated KAFKA-1661:
-
Labels: newbie test  (was: test)

> Move MockConsumer and MockProducer from src/main to src/test
> 
>
> Key: KAFKA-1661
> URL: https://issues.apache.org/jira/browse/KAFKA-1661
> Project: Kafka
>  Issue Type: Task
>  Components: clients, consumer, producer 
>Affects Versions: 0.8.1.1
> Environment: N/A
>Reporter: Andras Hatvani
>Assignee: Neha Narkhede
>Priority: Trivial
>  Labels: newbie, test
> Fix For: 0.8.2
>
>
> The MockConsumer and MockProducer are currently in src/main although they 
> belong in src/test.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1661) Move MockConsumer and MockProducer from src/main to src/test

[Neha Narkhede (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1661?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neha Narkhede updated KAFKA-1661:
-
Reviewer: Neha Narkhede
Assignee: (was: Neha Narkhede)

[~Andras Hatvani] Feel free to throw a patch. I can help you review and check 
it in.

> Move MockConsumer and MockProducer from src/main to src/test
> 
>
> Key: KAFKA-1661
> URL: https://issues.apache.org/jira/browse/KAFKA-1661
> Project: Kafka
>  Issue Type: Task
>  Components: clients, consumer, producer 
>Affects Versions: 0.8.1.1
> Environment: N/A
>Reporter: Andras Hatvani
>Priority: Trivial
>  Labels: newbie, test
> Fix For: 0.8.2
>
>
> The MockConsumer and MockProducer are currently in src/main although they 
> belong in src/test.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1558) AdminUtils.deleteTopic does not work

[Neha Narkhede (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1558?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154862#comment-14154862
 ] 

Neha Narkhede commented on KAFKA-1558:
--

[~sriharsha] Controller's inability to not shutdown on a soft failure is 
actually a problem and we need to look into it before fixing that case for 
delete topic. Would you mind filing another JIRA and stating the steps to 
reproduce the controller issue and make this depend on that JIRA? Since this 
JIRA is a blocker for 0.8.2, we are hoping to make progress on it. 

> AdminUtils.deleteTopic does not work
> 
>
> Key: KAFKA-1558
> URL: https://issues.apache.org/jira/browse/KAFKA-1558
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.8.1.1
>Reporter: Henning Schmiedehausen
>Assignee: Sriharsha Chintalapani
>Priority: Blocker
> Fix For: 0.8.2
>
> Attachments: kafka-thread-dump.log
>
>
> the AdminUtils:.deleteTopic method is implemented as
> {code}
> def deleteTopic(zkClient: ZkClient, topic: String) {
> ZkUtils.createPersistentPath(zkClient, 
> ZkUtils.getDeleteTopicPath(topic))
> }
> {code}
> but the DeleteTopicCommand actually does
> {code}
> zkClient = new ZkClient(zkConnect, 3, 3, ZKStringSerializer)
> zkClient.deleteRecursive(ZkUtils.getTopicPath(topic))
> {code}
> so I guess, that the 'createPersistentPath' above should actually be 
> {code}
> def deleteTopic(zkClient: ZkClient, topic: String) {
> ZkUtils.deletePathRecursive(zkClient, ZkUtils.getTopicPath(topic))
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1634) Update protocol wiki to reflect the new offset management feature

[Neha Narkhede (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1634?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154867#comment-14154867
 ] 

Neha Narkhede commented on KAFKA-1634:
--

[~jjkoshy] Thanks for the explanation. If retention is the use case, it is 
easier to understand if it was called retentionPeriod. Since 0.8.2 is when this 
releases, it will be great to make that renaming change before 0.8.2 goes out. 
If there is a broker side feature that is required for this to work properly 
and it's not done, then we can either get rid of this and add it when it's 
useful or leave it in there as a no-op or with a warning. What do you prefer?

> Update protocol wiki to reflect the new offset management feature
> -
>
> Key: KAFKA-1634
> URL: https://issues.apache.org/jira/browse/KAFKA-1634
> Project: Kafka
>  Issue Type: Bug
>Reporter: Neha Narkhede
>Assignee: Jun Rao
>Priority: Blocker
> Fix For: 0.8.2
>
>
> From the mailing list -
> following up on this -- I think the online API docs for OffsetCommitRequest
> still incorrectly refer to client-side timestamps:
> https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-OffsetCommitRequest
> Wasn't that removed and now always handled server-side now?  Would one of
> the devs mind updating the API spec wiki?



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: LeaderNotAvailableException, although leader elected

[Neha Narkhede]

Andras,

Thanks for your feedback!

In my opinion programmatic message sending must work out of the box on the
first try, without any exceptions, warnings or the need for additional
configuration.

I'd be glad to support/contribute.

I agree that the behavior of the producer for the first message on a topic
is awkward and I'd encourage feedback from you. We certainly are interested
in improving user experience. Could you please file a JIRA so we can
discuss alternatives there?

Thanks,
Neha

On Sat, Sep 27, 2014 at 12:58 AM, Andras Hatvani <
andras.hatv...@andrashatvani.com> wrote:

> AFAIK not topics, but only partitions of topics have leaders. What
> controllers do you mean? I haven't read about such.
>
> Thanks for the explanation regarding the metadata request, in the meantime
> I found out that this is an expected (!) failure (
> http://qnalist.com/questions/4787268/kafka-common-leadernotavailableexception
> ).
>
> For me this is isn't an acceptable way to communicate that the leader
> election is in progress. There is not a single hint to this fact, but only
> an exception.
> If this is an expected behavior, then it not only mustn't be an exception,
> but it also has to be communicated that there is something in progress.
> Furthermore, suggestions regarding changing the values variables I
> mentioned in my solution should be mandatory.
>
> This was my case:
> - OK, let's use Kafka
> - Create an infrastructure
> - Create a programmatic producer
> - Send a message
> - Message sending fails.
> - Retry
> - Message sending works!
> - Look for answers on the internet and in the docs
> - Read the configuration
> - Play around with configuration values.
>
> This is bad user experience especially for a newbie and involves a lot of
> effort.
>
> In my opinion programmatic message sending must work out of the box on the
> first try, without any exceptions, warnings or the need for additional
> configuration.
>
> I'd be glad to support/contribute.
>
> Regards,
> Andras
>
> > On 26 Sep 2014, at 19:53, Joel Koshy  wrote:
> >
> >>> kafka2_1  | [2014-09-26 12:35:07,289] INFO [Kafka Server 2],
> started (kafka.server.KafkaServer)
> >>> kafka2_1  | [2014-09-26 12:35:07,394] INFO New leader is 2
> (kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
> >
> > The above logs are for controller election. Not leader election for
> > the topic you are producing to.
> >
> > When you producer starts it will issue a metadata request and it
> > should auto-create the topic (provided auto-create is on which is
> > default). The first metadata request for a non-existent topic always
> > returns LeaderNotAvailable because the controller then has to elect a
> > leader for the new topic.
> >
> > Joel
> >
> > On Fri, Sep 26, 2014 at 04:07:58PM +0200, Andras Hatvani wrote:
> >> And the solution was:
> >> Increase retry.backoff.ms from the default 100 ms to 1000 ms, so the
> output is:
> >>
> >> 11891 [main] INFO  kafka.client.ClientUtils$ - Fetching metadata from
> broker id:0,host:192.168.59.103,port:9092 with correlation id 0 for 1
> topic(s) Set(inputTopic)
> >> 11893 [main] INFO  kafka.producer.SyncProducer - Connected to
> 192.168.59.103:9092 for producing
> >> 12045 [main] INFO  kafka.producer.SyncProducer - Disconnecting from
> 192.168.59.103:9092
> >> 12062 [main] WARN  kafka.producer.BrokerPartitionInfo - Error while
> fetching metadata [{TopicMetadata for topic inputTopic ->
> >> No partition metadata for topic inputTopic due to
> kafka.common.LeaderNotAvailableException}] for topic [inputTopic]: class
> kafka.common.LeaderNotAvailableException
> >> 12066 [main] INFO  kafka.client.ClientUtils$ - Fetching metadata from
> broker id:0,host:192.168.59.103,port:9092 with correlation id 1 for 1
> topic(s) Set(inputTopic)
> >> 12067 [main] INFO  kafka.producer.SyncProducer - Connected to
> 192.168.59.103:9092 for producing
> >> 12097 [main] INFO  kafka.producer.SyncProducer - Disconnecting from
> 192.168.59.103:9092
> >> 12098 [main] WARN  kafka.producer.BrokerPartitionInfo - Error while
> fetching metadata [{TopicMetadata for topic inputTopic ->
> >> No partition metadata for topic inputTopic due to
> kafka.common.LeaderNotAvailableException}] for topic [inputTopic]: class
> kafka.common.LeaderNotAvailableException
> >> 12098 [main] ERROR kafka.producer.async.DefaultEventHandler - Failed to
> collate messages by topic, partition due to: Failed to fetch topic metadata
> for topic: inputTopic
> >> 12099 [main] INFO  kafka.producer.async.DefaultEventHandler - Back off
> for 1000 ms before retrying send. Remaining retries = 3
> >> 13104 [main] INFO  kafka.client.ClientUtils$ - Fetching metadata from
> broker id:0,host:192.168.59.103,port:9092 with correlation id 2 for 1
> topic(s) Set(inputTopic)
> >> 13111 [main] INFO  kafka.producer.SyncProducer - Connected to
> 192.168.59.103:9092 for producing
> >> 13137 [main] INFO  kafka.producer.SyncProducer - Disconnecting from
> 192.168.59.103:9092
> >> 13161 [main] INFO

[jira] [Commented] (KAFKA-1558) AdminUtils.deleteTopic does not work

[Sriharsha Chintalapani (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1558?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154870#comment-14154870
 ] 

Sriharsha Chintalapani commented on KAFKA-1558:
---

[~nehanarkhede] working on finding the issue behind controller's shutdown. I'll 
file a JIRA.

> AdminUtils.deleteTopic does not work
> 
>
> Key: KAFKA-1558
> URL: https://issues.apache.org/jira/browse/KAFKA-1558
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.8.1.1
>Reporter: Henning Schmiedehausen
>Assignee: Sriharsha Chintalapani
>Priority: Blocker
> Fix For: 0.8.2
>
> Attachments: kafka-thread-dump.log
>
>
> the AdminUtils:.deleteTopic method is implemented as
> {code}
> def deleteTopic(zkClient: ZkClient, topic: String) {
> ZkUtils.createPersistentPath(zkClient, 
> ZkUtils.getDeleteTopicPath(topic))
> }
> {code}
> but the DeleteTopicCommand actually does
> {code}
> zkClient = new ZkClient(zkConnect, 3, 3, ZKStringSerializer)
> zkClient.deleteRecursive(ZkUtils.getTopicPath(topic))
> {code}
> so I guess, that the 'createPersistentPath' above should actually be 
> {code}
> def deleteTopic(zkClient: ZkClient, topic: String) {
> ZkUtils.deletePathRecursive(zkClient, ZkUtils.getTopicPath(topic))
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1661) Move MockConsumer and MockProducer from src/main to src/test

[Jay Kreps (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1661?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154877#comment-14154877
 ] 

Jay Kreps commented on KAFKA-1661:
--

I guess the thinking was that test/ was for *our* tests. No one should depend 
on our test code, it is just for us and has no external guarantees. The mocks 
on the other hand are part of what we deliver to users so from our perspective 
they are main not test. Also as a practical matter I really wanted them to show 
up in the main javadocs so that they would get documented and be discoverable.

Does that seem crazy/wrong?

> Move MockConsumer and MockProducer from src/main to src/test
> 
>
> Key: KAFKA-1661
> URL: https://issues.apache.org/jira/browse/KAFKA-1661
> Project: Kafka
>  Issue Type: Task
>  Components: clients, consumer, producer 
>Affects Versions: 0.8.1.1
> Environment: N/A
>Reporter: Andras Hatvani
>Priority: Trivial
>  Labels: newbie, test
> Fix For: 0.8.2
>
>
> The MockConsumer and MockProducer are currently in src/main although they 
> belong in src/test.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1660) Ability to call close() with a timeout on the Java Kafka Producer.

[Jay Kreps (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1660?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14154882#comment-14154882
 ] 

Jay Kreps commented on KAFKA-1660:
--

Gotcha, so if I understand close(time) is effectively tryClose(time) and 
attempts to close and then gives up if it can't, and abort is effectively 
forceClose() and forcefully kills the I/O thread.

> Ability to call close() with a timeout on the Java Kafka Producer. 
> ---
>
> Key: KAFKA-1660
> URL: https://issues.apache.org/jira/browse/KAFKA-1660
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, producer 
>Affects Versions: 0.8.2
>Reporter: Andrew Stein
>Assignee: Jun Rao
> Fix For: 0.8.2
>
>
> I would like the ability to call {{close}} with a timeout on the Java 
> Client's KafkaProducer.
> h6. Workaround
> Currently, it is possible to ensure that {{close}} will return quickly by 
> first doing a {{future.get(timeout)}} on the last future produced on each 
> partition, but this means that the user has to define the partitions up front 
> at the time of {{send}} and track the returned {{future}}'s



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-1663) Controller unable to shutdown after a soft failure

[Sriharsha Chintalapani (JIRA)]

Sriharsha Chintalapani created KAFKA-1663:
-

 Summary: Controller unable to shutdown after a soft failure
 Key: KAFKA-1663
 URL: https://issues.apache.org/jira/browse/KAFKA-1663
 Project: Kafka
  Issue Type: Bug
Reporter: Sriharsha Chintalapani


As part of testing KAFKA-1558 I came across a case where inducing soft failure 
in the current controller elects a new controller  but the old controller 
doesn't shutdown properly.
steps to reproduce
1) 5 broker cluster
2) high number of topics(I tested it with 1000 topics)
3) on the current controller do kill -SIGSTOP  pid( broker's process id)
4) wait for bit over zookeeper timeout (server.properties)
5) kill -SIGCONT pid
6) There will be a new controller elected. check old controller's
log 
[2014-09-30 15:59:53,398] INFO [SessionExpirationListener on 1], ZK expired; 
shut down all controller components and try to re-elect 
(kafka.controller.KafkaController$SessionExpirationListener)
[2014-09-30 15:59:53,400] INFO [delete-topics-thread-1], Shutting down 
(kafka.controller.TopicDeletionManager$DeleteTopicsThread)

If it stops there and the broker  logs keeps printing 
Cached zkVersion [0] not equal to that in zookeeper, skip updating ISR 
(kafka.cluster.Partition)
than the controller shutdown never completes.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1663) Controller unable to shutdown after a soft failure

[Neha Narkhede (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1663?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neha Narkhede updated KAFKA-1663:
-
Fix Version/s: 0.8.2

> Controller unable to shutdown after a soft failure
> --
>
> Key: KAFKA-1663
> URL: https://issues.apache.org/jira/browse/KAFKA-1663
> Project: Kafka
>  Issue Type: Bug
>Reporter: Sriharsha Chintalapani
>Priority: Blocker
> Fix For: 0.8.2
>
>
> As part of testing KAFKA-1558 I came across a case where inducing soft 
> failure in the current controller elects a new controller  but the old 
> controller doesn't shutdown properly.
> steps to reproduce
> 1) 5 broker cluster
> 2) high number of topics(I tested it with 1000 topics)
> 3) on the current controller do kill -SIGSTOP  pid( broker's process id)
> 4) wait for bit over zookeeper timeout (server.properties)
> 5) kill -SIGCONT pid
> 6) There will be a new controller elected. check old controller's
> log 
> [2014-09-30 15:59:53,398] INFO [SessionExpirationListener on 1], ZK expired; 
> shut down all controller components and try to re-elect 
> (kafka.controller.KafkaController$SessionExpirationListener)
> [2014-09-30 15:59:53,400] INFO [delete-topics-thread-1], Shutting down 
> (kafka.controller.TopicDeletionManager$DeleteTopicsThread)
> If it stops there and the broker  logs keeps printing 
> Cached zkVersion [0] not equal to that in zookeeper, skip updating ISR 
> (kafka.cluster.Partition)
> than the controller shutdown never completes.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1663) Controller unable to shutdown after a soft failure

[Neha Narkhede (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1663?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neha Narkhede updated KAFKA-1663:
-
Priority: Blocker  (was: Major)

> Controller unable to shutdown after a soft failure
> --
>
> Key: KAFKA-1663
> URL: https://issues.apache.org/jira/browse/KAFKA-1663
> Project: Kafka
>  Issue Type: Bug
>Reporter: Sriharsha Chintalapani
>Priority: Blocker
> Fix For: 0.8.2
>
>
> As part of testing KAFKA-1558 I came across a case where inducing soft 
> failure in the current controller elects a new controller  but the old 
> controller doesn't shutdown properly.
> steps to reproduce
> 1) 5 broker cluster
> 2) high number of topics(I tested it with 1000 topics)
> 3) on the current controller do kill -SIGSTOP  pid( broker's process id)
> 4) wait for bit over zookeeper timeout (server.properties)
> 5) kill -SIGCONT pid
> 6) There will be a new controller elected. check old controller's
> log 
> [2014-09-30 15:59:53,398] INFO [SessionExpirationListener on 1], ZK expired; 
> shut down all controller components and try to re-elect 
> (kafka.controller.KafkaController$SessionExpirationListener)
> [2014-09-30 15:59:53,400] INFO [delete-topics-thread-1], Shutting down 
> (kafka.controller.TopicDeletionManager$DeleteTopicsThread)
> If it stops there and the broker  logs keeps printing 
> Cached zkVersion [0] not equal to that in zookeeper, skip updating ISR 
> (kafka.cluster.Partition)
> than the controller shutdown never completes.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1662) gradle release issue permgen space

[Joe Stein (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1662?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Joe Stein updated KAFKA-1662:
-
Description: 
Finding issues doing the kafka release with permgen space

./gradlew releaseTarGzAll

ant:scaladoc] java.lang.OutOfMemoryError: PermGen space
:kafka-0.8.2-ALPHA1-src:core:scaladoc FAILED
:releaseTarGz_2_10_1 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Execution failed for task ':core:scaladoc'.
> PermGen space

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 5 mins 55.53 secs

FAILURE: Build failed with an exception.

* What went wrong:
PermGen space



  was:
Finding issues doing the kafka release with permgen space

ant:scaladoc] java.lang.OutOfMemoryError: PermGen space
:kafka-0.8.2-ALPHA1-src:core:scaladoc FAILED
:releaseTarGz_2_10_1 FAILED

FAILURE: Build failed with an exception.

* What went wrong:
Execution failed for task ':core:scaladoc'.
> PermGen space

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or --debug 
option to get more log output.

BUILD FAILED

Total time: 5 mins 55.53 secs

FAILURE: Build failed with an exception.

* What went wrong:
PermGen space




> gradle release issue permgen space
> --
>
> Key: KAFKA-1662
> URL: https://issues.apache.org/jira/browse/KAFKA-1662
> Project: Kafka
>  Issue Type: Bug
>Reporter: Joe Stein
>Priority: Blocker
>  Labels: newbie
> Fix For: 0.8.2
>
>
> Finding issues doing the kafka release with permgen space
> ./gradlew releaseTarGzAll
> ant:scaladoc] java.lang.OutOfMemoryError: PermGen space
> :kafka-0.8.2-ALPHA1-src:core:scaladoc FAILED
> :releaseTarGz_2_10_1 FAILED
> FAILURE: Build failed with an exception.
> * What went wrong:
> Execution failed for task ':core:scaladoc'.
> > PermGen space
> * Try:
> Run with --stacktrace option to get the stack trace. Run with --info or 
> --debug option to get more log output.
> BUILD FAILED
> Total time: 5 mins 55.53 secs
> FAILURE: Build failed with an exception.
> * What went wrong:
> PermGen space



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: Two open issues on Kafka security

[Jonathan Creasy]

This is not nearly as deep as the discussion so far, but I did want to
throw this idea out there to make sure we¹ve thought about it.

The Kafka project should make sure that when deployed alongside a Hadoop
cluster from any major distributions that it can tie seamlessly into the
authentication and authorization used within that cluster. For example,
Apache Sentry.

This may present additional difficulties that means a decision is made to
not do that or alternatively the Kerberos authentication and the
authorization schemes we are already working on may be sufficient.

I¹m not sure that anything I¹ve read so far in this discussion actually
poses a problem, but I¹m an Ops guy and being able to more easily
integrate more things, makes my life better. :)

-Jonathan

On 9/30/14, 11:26 PM, "Joe Stein"  wrote:

>inline
>
>On Tue, Sep 30, 2014 at 11:58 PM, Jay Kreps  wrote:
>
>> Hey Joe,
>>
>> For (1) what are you thinking for the PermissionManager api?
>>
>> The way I see it, the first question we have to answer is whether it
>> is possible to make authentication and authorization independent. What
>> I mean by that is whether I can write an authorization library that
>> will work the same whether you authenticate with ssl or kerberos.
>
>
>To me that is a requirement. We can't tie them together.  We have to
>provide the ability for authorization to work regardless of the
>authentication.  One *VERY* important use case is level of trust in
>authentication from the authorization perpsective.  e.g. I authorize
>"identity" based on the how you authenticated Alice is able to view
>topic X if Alice authenticated over kerberos.  Bob isn't allowed to view
>topic X no matter what. Alice can authenticate over not kerberos (uses
>cases for that) and in that case Alice wouldn't see topic X.  A concrete
>use case for this with Kafka would be a third party bank consuming data to
>a broker.  The service provider would have some kerberos local auth for
>that bank to-do back up that would also have access to other topics
>related
>to that banks data the bank itself over SSL wants a stream of events
>(some specific topic) and that banks identity only sees that topic.  It is
>important to not confuse identity, authentication and authorization.
>
>
>> If
>> so then we need to pick some subset of identity information that we
>> can extract from both and have this constitute the identity we pass
>> into the authorization interface. The original proposal had just the
>> username/subject. But maybe we should add the ip address as well as
>> that is useful. What I would prefer not to do is add everything in the
>> certificate. I think the assumption is that you are generating these
>> certificates for Kafka so you can put whatever identity info you want
>> in the Subject Alternative Name. If that is true then just using that
>> should be okay, right?
>>
>
>I think we should just push the byte[] and let the plugin deal with it.
>So, if we have a certificate object then pass that along with whatever
>other meta data (e.g. IP address of client) we can.  I don't think we
>should do any parsing whatsover and let the plugin deal with that.  Any
>parsing we do on the identity information for the "security object" forces
>us into specific implementations and I don't see any reason to-do that...
>If plug-ins want an "easier" time to deal with certs and parsing and blah
>blah blah then we can implement some way they can do this without much
>fuss we also need to make sure that crypto library is plugable too (so
>we can expose an API for them to call) so that HSM can be easily dropped
>in
>without Kafka caring... so in the plugin we could provide a
>indentity.getAlternativeAttribute() and then that use case is solved (and
>we can use bouncy castle or whatever to parse it for them to make it
>easier) and always give them raw bytes so they could do it themselves.
>
>
>>
>> -Jay
>>
>>
>>
>>
>>
>> On Tue, Sep 30, 2014 at 4:09 PM, Joe Stein  wrote:
>> > 1) We need to support the most flexibility we can and make this
>> transparent
>> > to kafka (to use Gwen's term).  Any specific implementation is going
>>to
>> > make it not work with some solution stopping people from using Kafka.
>> That
>> > is a reality because everyone just does it slightly differently
>>enough.
>> If
>> > we have an "identity" byte structure (lets not use string because some
>> > security objects are bytes) this should just fall through to the
>> > implementor.  For certs this is the entire x509 object (not just the
>> > certificate part as it could contain an ASN.1 timestamp) and inside
>>you
>> > parse and do what you want with it.
>> >
>> > 2) While I think there are many benefits to just the handshake
>>approach I
>> > don't think it outweighs the cons Jay expressed. a) We can't lead the
>> > client libraries down a new path of interacting with Kafka.  By
>> > incrementally adding to the wire protocol we are directing a very
>>clear
>> and
>> > expect ted app

Re: Two open issues on Kafka security

[Joe Stein]

Hi Jonathan,

"Hadoop delegation tokens to enable MapReduce, Samza, or other frameworks
running in the Hadoop environment to access Kafka"
https://cwiki.apache.org/confluence/display/KAFKA/Security is on the list,
yup!

/***
 Joe Stein
 Founder, Principal Consultant
 Big Data Open Source Security LLC
 http://www.stealth.ly
 Twitter: @allthingshadoop 
/

On Wed, Oct 1, 2014 at 12:35 PM, Jonathan Creasy 
wrote:

> This is not nearly as deep as the discussion so far, but I did want to
> throw this idea out there to make sure we¹ve thought about it.
>
> The Kafka project should make sure that when deployed alongside a Hadoop
> cluster from any major distributions that it can tie seamlessly into the
> authentication and authorization used within that cluster. For example,
> Apache Sentry.
>
> This may present additional difficulties that means a decision is made to
> not do that or alternatively the Kerberos authentication and the
> authorization schemes we are already working on may be sufficient.
>
> I¹m not sure that anything I¹ve read so far in this discussion actually
> poses a problem, but I¹m an Ops guy and being able to more easily
> integrate more things, makes my life better. :)
>
> -Jonathan
>
> On 9/30/14, 11:26 PM, "Joe Stein"  wrote:
>
> >inline
> >
> >On Tue, Sep 30, 2014 at 11:58 PM, Jay Kreps  wrote:
> >
> >> Hey Joe,
> >>
> >> For (1) what are you thinking for the PermissionManager api?
> >>
> >> The way I see it, the first question we have to answer is whether it
> >> is possible to make authentication and authorization independent. What
> >> I mean by that is whether I can write an authorization library that
> >> will work the same whether you authenticate with ssl or kerberos.
> >
> >
> >To me that is a requirement. We can't tie them together.  We have to
> >provide the ability for authorization to work regardless of the
> >authentication.  One *VERY* important use case is level of trust in
> >authentication from the authorization perpsective.  e.g. I authorize
> >"identity" based on the how you authenticated Alice is able to view
> >topic X if Alice authenticated over kerberos.  Bob isn't allowed to view
> >topic X no matter what. Alice can authenticate over not kerberos (uses
> >cases for that) and in that case Alice wouldn't see topic X.  A concrete
> >use case for this with Kafka would be a third party bank consuming data to
> >a broker.  The service provider would have some kerberos local auth for
> >that bank to-do back up that would also have access to other topics
> >related
> >to that banks data the bank itself over SSL wants a stream of events
> >(some specific topic) and that banks identity only sees that topic.  It is
> >important to not confuse identity, authentication and authorization.
> >
> >
> >> If
> >> so then we need to pick some subset of identity information that we
> >> can extract from both and have this constitute the identity we pass
> >> into the authorization interface. The original proposal had just the
> >> username/subject. But maybe we should add the ip address as well as
> >> that is useful. What I would prefer not to do is add everything in the
> >> certificate. I think the assumption is that you are generating these
> >> certificates for Kafka so you can put whatever identity info you want
> >> in the Subject Alternative Name. If that is true then just using that
> >> should be okay, right?
> >>
> >
> >I think we should just push the byte[] and let the plugin deal with it.
> >So, if we have a certificate object then pass that along with whatever
> >other meta data (e.g. IP address of client) we can.  I don't think we
> >should do any parsing whatsover and let the plugin deal with that.  Any
> >parsing we do on the identity information for the "security object" forces
> >us into specific implementations and I don't see any reason to-do that...
> >If plug-ins want an "easier" time to deal with certs and parsing and blah
> >blah blah then we can implement some way they can do this without much
> >fuss we also need to make sure that crypto library is plugable too (so
> >we can expose an API for them to call) so that HSM can be easily dropped
> >in
> >without Kafka caring... so in the plugin we could provide a
> >indentity.getAlternativeAttribute() and then that use case is solved (and
> >we can use bouncy castle or whatever to parse it for them to make it
> >easier) and always give them raw bytes so they could do it themselves.
> >
> >
> >>
> >> -Jay
> >>
> >>
> >>
> >>
> >>
> >> On Tue, Sep 30, 2014 at 4:09 PM, Joe Stein 
> wrote:
> >> > 1) We need to support the most flexibility we can and make this
> >> transparent
> >> > to kafka (to use Gwen's term).  Any specific implementation is going
> >>to
> >> > make it not work with some solution stopping people from using Kafka.
> >> That
> >> > is a reality because everyone j

[jira] [Commented] (KAFKA-1494) Failed to send messages after 3 tries.

[Josh Rosen (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1494?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155113#comment-14155113
 ] 

Josh Rosen commented on KAFKA-1494:
---

Hi Joe,

Sorry for cluttering up this JIRA; I should have posted to the mailing list.  I 
was _just_ about to hit 'send' on my message when I tried one more thing and 
stumbled on a solution for my issue:

It turns out that publishing to auto-created topics seems to work in my setup 
(using the configurations tweaks suggested here); it was only publication to 
manually-created topics that was failing.  The root problem turned out to be 
that I had created ZkClient without specifying a serializer, which seemed to 
allow certain ZK metadata operations to succeed even though the stored data 
didn't seem to be usable by the broker / producer.  When I configured ZkClient 
to use ZKStringSerializer, the problem disappeared and I could publish data as 
expected.  Hope this helps anyone who has run into a similar issue (and sorry 
for the clutter!).

> Failed to send messages after 3 tries.
> --
>
> Key: KAFKA-1494
> URL: https://issues.apache.org/jira/browse/KAFKA-1494
> Project: Kafka
>  Issue Type: Bug
>  Components: controller, core
>Affects Versions: 0.8.1.1
> Environment: Mac OS 
>Reporter: darion yaphets
>Assignee: Neha Narkhede
>
> I use default server & zookeeper config to start-up zookeeper server and 
> kafka broker on my machine to test custom message which based on proto buffer 
> . I write a client to send protobuf-message to kafka broker and source code 
> as following :
> Properties properties = new Properties();
>   properties.put("serializer.class", 
> "java_example.ProtoBufMessage");
>   properties.put("metadata.broker.list", "localhost:9092");
>   ProducerConfig config = new ProducerConfig(properties);
>   testBuf buffer = testBuf.newBuilder().setID(0)
>   .setUrl("darion.yaphet.org").build();
>   Producer producer = new Producer testBuf>(
>   config);
>   producer.send(new KeyedMessage("protobuffer", 
> buffer));
> client debug log report a exception:
> [FileSystemMoniter] INFO [main] kafka.utils.Logging$class.info(68) | 
> Disconnecting from localhost:9092
> [FileSystemMoniter] DEBUG [main] kafka.utils.Logging$class.debug(52) | 
> Successfully fetched metadata for 1 topic(s) Set(protobuffer)
> [FileSystemMoniter] WARN [main] kafka.utils.Logging$class.warn(83) | Error 
> while fetching metadata [{TopicMetadata for topic protobuffer -> 
> No partition metadata for topic protobuffer due to 
> kafka.common.LeaderNotAvailableException}] for topic [protobuffer]: class 
> kafka.common.LeaderNotAvailableException 
> [FileSystemMoniter] ERROR [main] kafka.utils.Logging$class.error(97) | Failed 
> to send requests for topics protobuffer with correlation ids in [0,8]
> Exception in thread "main" kafka.common.FailedToSendMessageException: Failed 
> to send messages after 3 tries.
>   at 
> kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
>   at kafka.producer.Producer.send(Producer.scala:76)
>   at kafka.javaapi.producer.Producer.send(Producer.scala:33)
>   at java_example.ProducerExamples.main(ProducerExamples.java:26)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1494) Failed to send messages after 3 tries.

[Joe Stein (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1494?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155148#comment-14155148
 ] 

Joe Stein commented on KAFKA-1494:
--

[~joshrosen] no worries about the clutter, it is just that more people are on 
the mailing list so you get more information/feedback from a larger group.  
Thanks for posting for posterity for folks, it is helpful for other that come 
along.

> Failed to send messages after 3 tries.
> --
>
> Key: KAFKA-1494
> URL: https://issues.apache.org/jira/browse/KAFKA-1494
> Project: Kafka
>  Issue Type: Bug
>  Components: controller, core
>Affects Versions: 0.8.1.1
> Environment: Mac OS 
>Reporter: darion yaphets
>Assignee: Neha Narkhede
>
> I use default server & zookeeper config to start-up zookeeper server and 
> kafka broker on my machine to test custom message which based on proto buffer 
> . I write a client to send protobuf-message to kafka broker and source code 
> as following :
> Properties properties = new Properties();
>   properties.put("serializer.class", 
> "java_example.ProtoBufMessage");
>   properties.put("metadata.broker.list", "localhost:9092");
>   ProducerConfig config = new ProducerConfig(properties);
>   testBuf buffer = testBuf.newBuilder().setID(0)
>   .setUrl("darion.yaphet.org").build();
>   Producer producer = new Producer testBuf>(
>   config);
>   producer.send(new KeyedMessage("protobuffer", 
> buffer));
> client debug log report a exception:
> [FileSystemMoniter] INFO [main] kafka.utils.Logging$class.info(68) | 
> Disconnecting from localhost:9092
> [FileSystemMoniter] DEBUG [main] kafka.utils.Logging$class.debug(52) | 
> Successfully fetched metadata for 1 topic(s) Set(protobuffer)
> [FileSystemMoniter] WARN [main] kafka.utils.Logging$class.warn(83) | Error 
> while fetching metadata [{TopicMetadata for topic protobuffer -> 
> No partition metadata for topic protobuffer due to 
> kafka.common.LeaderNotAvailableException}] for topic [protobuffer]: class 
> kafka.common.LeaderNotAvailableException 
> [FileSystemMoniter] ERROR [main] kafka.utils.Logging$class.error(97) | Failed 
> to send requests for topics protobuffer with correlation ids in [0,8]
> Exception in thread "main" kafka.common.FailedToSendMessageException: Failed 
> to send messages after 3 tries.
>   at 
> kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90)
>   at kafka.producer.Producer.send(Producer.scala:76)
>   at kafka.javaapi.producer.Producer.send(Producer.scala:33)
>   at java_example.ProducerExamples.main(ProducerExamples.java:26)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: Two open issues on Kafka security

[Jarek Jarcec Cecho]

I’m following the security proposal wiki page [1] and this discussion and I 
would like to jump in with few points if I might :)  Let me start by saying 
that I like the material and the discussion here, good work!

I was part of the team who originally designed and worked on Sentry and I 
wanted to share few to see how it will resonate with people.  My first and 
probably biggest point would be to separate authorization and authentication as 
two separate systems. I believe that Jao has already stressed that in the email 
thread, but I wanted to reiterate on that point. In my experience users don’t 
care that much about how the user has been authenticated if they trust that 
mechanism, what they care more about is that the authorization model is 
consistent and behaves the same way. E.g. if I configured that user jarcec can 
write into topic “logs”, he should be able to do that no matter where the 
connection came from - whether he has been authorized from Kerberos as he is 
directly exploring the data from his computer, he is authorized through 
delegation token because he is running map reduce jobs calculating statistics 
or he is  authorized through SSL certificated because … (well I’m missing good 
example here, but you’re probably following my point).

I’ve also noticed that we are planning to have no hierarchy in the authz object 
model per the wiki [1] with the reasoning that Kafka do not supports topic 
hierarchy. I see that point, but at the same time it got me thinking - are we 
sure that Kafka will never have hierarchic topics? Seems as a nice feature that 
might be usable for some use cases and something that we might want to add in 
the future. But regardless of that I would suggest to introduce a hierarchy 
anyway, even though if it would be just two levels. In sentry (for Hive) we’ve 
introduced concept of “Service” where all the databases are children of the 
service. In Kafka I would imagine that we would have “service” and “topics” as 
the children. Having this is much easier to model general privileges where you 
need to grant access to all topics - you will just grant access to the entire 
service and all topics will get “inherited”.

I’m wondering what are other people thoughts?

Jarcec

Links:
1: https://cwiki.apache.org/confluence/display/KAFKA/Security

On Oct 1, 2014, at 9:44 AM, Joe Stein  wrote:

> Hi Jonathan,
> 
> "Hadoop delegation tokens to enable MapReduce, Samza, or other frameworks
> running in the Hadoop environment to access Kafka"
> https://cwiki.apache.org/confluence/display/KAFKA/Security is on the list,
> yup!
> 
> /***
> Joe Stein
> Founder, Principal Consultant
> Big Data Open Source Security LLC
> http://www.stealth.ly
> Twitter: @allthingshadoop 
> /
> 
> On Wed, Oct 1, 2014 at 12:35 PM, Jonathan Creasy 
> wrote:
> 
>> This is not nearly as deep as the discussion so far, but I did want to
>> throw this idea out there to make sure we¹ve thought about it.
>> 
>> The Kafka project should make sure that when deployed alongside a Hadoop
>> cluster from any major distributions that it can tie seamlessly into the
>> authentication and authorization used within that cluster. For example,
>> Apache Sentry.
>> 
>> This may present additional difficulties that means a decision is made to
>> not do that or alternatively the Kerberos authentication and the
>> authorization schemes we are already working on may be sufficient.
>> 
>> I¹m not sure that anything I¹ve read so far in this discussion actually
>> poses a problem, but I¹m an Ops guy and being able to more easily
>> integrate more things, makes my life better. :)
>> 
>> -Jonathan
>> 
>> On 9/30/14, 11:26 PM, "Joe Stein"  wrote:
>> 
>>> inline
>>> 
>>> On Tue, Sep 30, 2014 at 11:58 PM, Jay Kreps  wrote:
>>> 
 Hey Joe,
 
 For (1) what are you thinking for the PermissionManager api?
 
 The way I see it, the first question we have to answer is whether it
 is possible to make authentication and authorization independent. What
 I mean by that is whether I can write an authorization library that
 will work the same whether you authenticate with ssl or kerberos.
>>> 
>>> 
>>> To me that is a requirement. We can't tie them together.  We have to
>>> provide the ability for authorization to work regardless of the
>>> authentication.  One *VERY* important use case is level of trust in
>>> authentication from the authorization perpsective.  e.g. I authorize
>>> "identity" based on the how you authenticated Alice is able to view
>>> topic X if Alice authenticated over kerberos.  Bob isn't allowed to view
>>> topic X no matter what. Alice can authenticate over not kerberos (uses
>>> cases for that) and in that case Alice wouldn't see topic X.  A concrete
>>> use case for this with Kafka would be a third party bank consuming data to
>>> a broker.  The service provider would ha

Re: Two open issues on Kafka security

[Michael Herstine]

Regarding question #1, I’m not sure I follow you, Joe: you’re proposing (I
think) that the API take a byte[], but what will be in that array? A
serialized certificate if the client authenticated via SSL and the
principal name (perhaps normalized) if the client authenticated via
Kerberos?

Regarding question #2, I think I was unclear in the meeting yesterday: I
was proposing a separate port for each authentication method (including
none). That is, if a client wants no authentication, then they would
connect to port N on the broker. If they wanted to talk over SSL, then
they connect to port N+1 (say). Kerberos: N+2. This would remove the need
for a new request, since the authentication type would be implicit in the
port on which the client connected (and it was my understanding that it
was desirable to not introduce any new messages).

Perhaps the confusion comes from the fact, correctly pointed out by Jay,
that when you want to use SASL on a single port, there does of course need
to be a way for the incoming client to signal which mechanism it wants to
use, and that’s out of scope of the SASL spec. I didn’t see there being a
desire to add new SASL mechanisms going forward, but perhaps I was
incorrect?

In any event, I’d like to suggest we keep the “open” or “no auth” port
separate, both to make it easy for admins to force the use of security (by
shutting down that port) and to avoid downgrade attacks (where an attacker
intercepts the opening packet from a client requesting security & alters
it to request none).

I’ll update the Wiki with my notes from yesterday’s meeting this afternoon.

Thanks,

On 10/1/14, 9:35 AM, "Jonathan Creasy"  wrote:

>This is not nearly as deep as the discussion so far, but I did want to
>throw this idea out there to make sure we¹ve thought about it.
>
>The Kafka project should make sure that when deployed alongside a Hadoop
>cluster from any major distributions that it can tie seamlessly into the
>authentication and authorization used within that cluster. For example,
>Apache Sentry.
>
>This may present additional difficulties that means a decision is made to
>not do that or alternatively the Kerberos authentication and the
>authorization schemes we are already working on may be sufficient.
>
>I¹m not sure that anything I¹ve read so far in this discussion actually
>poses a problem, but I¹m an Ops guy and being able to more easily
>integrate more things, makes my life better. :)
>
>-Jonathan
>
>On 9/30/14, 11:26 PM, "Joe Stein"  wrote:
>
>>inline
>>
>>On Tue, Sep 30, 2014 at 11:58 PM, Jay Kreps  wrote:
>>
>>> Hey Joe,
>>>
>>> For (1) what are you thinking for the PermissionManager api?
>>>
>>> The way I see it, the first question we have to answer is whether it
>>> is possible to make authentication and authorization independent. What
>>> I mean by that is whether I can write an authorization library that
>>> will work the same whether you authenticate with ssl or kerberos.
>>
>>
>>To me that is a requirement. We can't tie them together.  We have to
>>provide the ability for authorization to work regardless of the
>>authentication.  One *VERY* important use case is level of trust in
>>authentication from the authorization perpsective.  e.g. I authorize
>>"identity" based on the how you authenticated Alice is able to view
>>topic X if Alice authenticated over kerberos.  Bob isn't allowed to view
>>topic X no matter what. Alice can authenticate over not kerberos (uses
>>cases for that) and in that case Alice wouldn't see topic X.  A concrete
>>use case for this with Kafka would be a third party bank consuming data
>>to
>>a broker.  The service provider would have some kerberos local auth for
>>that bank to-do back up that would also have access to other topics
>>related
>>to that banks data the bank itself over SSL wants a stream of events
>>(some specific topic) and that banks identity only sees that topic.  It
>>is
>>important to not confuse identity, authentication and authorization.
>>
>>
>>> If
>>> so then we need to pick some subset of identity information that we
>>> can extract from both and have this constitute the identity we pass
>>> into the authorization interface. The original proposal had just the
>>> username/subject. But maybe we should add the ip address as well as
>>> that is useful. What I would prefer not to do is add everything in the
>>> certificate. I think the assumption is that you are generating these
>>> certificates for Kafka so you can put whatever identity info you want
>>> in the Subject Alternative Name. If that is true then just using that
>>> should be okay, right?
>>>
>>
>>I think we should just push the byte[] and let the plugin deal with it.
>>So, if we have a certificate object then pass that along with whatever
>>other meta data (e.g. IP address of client) we can.  I don't think we
>>should do any parsing whatsover and let the plugin deal with that.  Any
>>parsing we do on the identity information for the "security object"
>>forces
>>u

[jira] [Commented] (KAFKA-1660) Ability to call close() with a timeout on the Java Kafka Producer.

[Andrew Stein (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1660?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155625#comment-14155625
 ] 

Andrew Stein commented on KAFKA-1660:
-

Yes, {{close(timeout)}} is {{tryClose(timeout)}}.

On the other hand {{abort()}} does not even need to try to {{forceClose()}}. It 
kills the I/O thread. hopefully in a clean manner, perhaps by posting an 
"abortFlag" to the sender. The sender can check for this flag in the two loops 
of its {{run()}} -- This is just a suggestion.


> Ability to call close() with a timeout on the Java Kafka Producer. 
> ---
>
> Key: KAFKA-1660
> URL: https://issues.apache.org/jira/browse/KAFKA-1660
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, producer 
>Affects Versions: 0.8.2
>Reporter: Andrew Stein
>Assignee: Jun Rao
> Fix For: 0.8.2
>
>
> I would like the ability to call {{close}} with a timeout on the Java 
> Client's KafkaProducer.
> h6. Workaround
> Currently, it is possible to ensure that {{close}} will return quickly by 
> first doing a {{future.get(timeout)}} on the last future produced on each 
> partition, but this means that the user has to define the partitions up front 
> at the time of {{send}} and track the returned {{future}}'s



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1590) Binarize trace level request logging along with debug level text logging

[Guozhang Wang (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1590?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155647#comment-14155647
 ] 

Guozhang Wang commented on KAFKA-1590:
--

[~jkreps] I agree that this would not be the default config. 

By default: the request log would be written in text.

When this knob turns on: the request log would be written only as summary in 
text (i.e. today's DEBUG level logging), and the detailed log will be recorded 
as kafka-request-details.gzip/[any other compression types?].

I personally do not know what the % savings would be with the proposed binary / 
compression format compared the the text format. Would be great if we can get 
some numbers with a preliminary patch.

[~Abhishek Sharma] do you think it is possible that we write a simple 
cross-platform tool to parse the request-details binary files?

> Binarize trace level request logging along with debug level text logging
> 
>
> Key: KAFKA-1590
> URL: https://issues.apache.org/jira/browse/KAFKA-1590
> Project: Kafka
>  Issue Type: Bug
>Reporter: Guozhang Wang
>Assignee: Abhishek Sharma
>  Labels: newbie
> Fix For: 0.9.0
>
> Attachments: Jira-1590-Log4j Binary Appender.patch
>
>
> With trace level logging, the request handling logs can grow very fast 
> depending on the client behavior (e.g. consumer with 0 maxWait and hence keep 
> sending fetch requests). Previously we have changed it to debug level which 
> only provides a summary of the requests, omitting request details. However 
> this does not work perfectly since summaries are not sufficient for 
> trouble-shooting, and turning on trace level upon issues will be too late.
> The proposed solution here, is to default to debug level logging with trace 
> level logging printed as binary format at the same time. The generated binary 
> files can then be further compressed / rolled out. When needed, we will then 
> decompress / parse the trace logs into texts.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-294) "Path length must be > 0" error during startup

[Gwen Shapira (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-294?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Gwen Shapira updated KAFKA-294:
---
Fix Version/s: 0.8.2

> "Path length must be > 0" error during startup
> --
>
> Key: KAFKA-294
> URL: https://issues.apache.org/jira/browse/KAFKA-294
> Project: Kafka
>  Issue Type: Bug
>Reporter: Thomas Dudziak
> Fix For: 0.8.2
>
>
> When starting Kafka 0.7.0 using zkclient-0.1.jar, I get this error:
> INFO 2012-03-06 02:39:04,072  main kafka.server.KafkaZooKeeper Registering 
> broker /brokers/ids/1
> FATAL 2012-03-06 02:39:04,111  main kafka.server.KafkaServer Fatal error 
> during startup.
> java.lang.IllegalArgumentException: Path length must be > 0
> at 
> org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
> at 
> org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
> at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:620)
> at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
> at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
> at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
> at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
> at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
> at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
> at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
> at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
> at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:48)
> at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:60)
> at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:72)
> at 
> kafka.server.KafkaZooKeeper.registerBrokerInZk(KafkaZooKeeper.scala:57)
> at kafka.log.LogManager.startup(LogManager.scala:124)
> at kafka.server.KafkaServer.startup(KafkaServer.scala:80)
> at 
> kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:47)
> at kafka.Kafka$.main(Kafka.scala:60)
> at kafka.Kafka.main(Kafka.scala)
> The problem seems to be this code in ZkClient's createPersistent method:
> String parentDir = path.substring(0, path.lastIndexOf('/'));
> createPersistent(parentDir, createParents);
> createPersistent(path, createParents);
> which doesn't check for whether parentDir is an empty string, which it will 
> become for /brokers/ids/1 after two recursions.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-294) "Path length must be > 0" error during startup

[Gwen Shapira (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-294?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Gwen Shapira resolved KAFKA-294.

Resolution: Fixed

With KAFKA-404 committed, this is resolved too.

> "Path length must be > 0" error during startup
> --
>
> Key: KAFKA-294
> URL: https://issues.apache.org/jira/browse/KAFKA-294
> Project: Kafka
>  Issue Type: Bug
>Reporter: Thomas Dudziak
>
> When starting Kafka 0.7.0 using zkclient-0.1.jar, I get this error:
> INFO 2012-03-06 02:39:04,072  main kafka.server.KafkaZooKeeper Registering 
> broker /brokers/ids/1
> FATAL 2012-03-06 02:39:04,111  main kafka.server.KafkaServer Fatal error 
> during startup.
> java.lang.IllegalArgumentException: Path length must be > 0
> at 
> org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
> at 
> org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
> at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:620)
> at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
> at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
> at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
> at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
> at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
> at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
> at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
> at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
> at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:48)
> at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:60)
> at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:72)
> at 
> kafka.server.KafkaZooKeeper.registerBrokerInZk(KafkaZooKeeper.scala:57)
> at kafka.log.LogManager.startup(LogManager.scala:124)
> at kafka.server.KafkaServer.startup(KafkaServer.scala:80)
> at 
> kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:47)
> at kafka.Kafka$.main(Kafka.scala:60)
> at kafka.Kafka.main(Kafka.scala)
> The problem seems to be this code in ZkClient's createPersistent method:
> String parentDir = path.substring(0, path.lastIndexOf('/'));
> createPersistent(parentDir, createParents);
> createPersistent(path, createParents);
> which doesn't check for whether parentDir is an empty string, which it will 
> become for /brokers/ids/1 after two recursions.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1644) Inherit FetchResponse from RequestOrResponse

[Guozhang Wang (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1644?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155678#comment-14155678
 ] 

Guozhang Wang commented on KAFKA-1644:
--

Anton, I think Jun's not saying that there is a blocker for doing so, it's just 
that since FetchResponse does not use writeTo to write to the socket, but use 
FetchReponseSend instead for zero-copy, it may not logically be treated as 
inheritance of RequestOrResponse. After looking into your patch I see you are 
enforcing this by letting FetchResponseSend to just call FetchResponse.writeTo 
instead, which I think is OK if we do have a strong reason for letting 
FetchResponse inheriting from RequestOrResponse.

> Inherit FetchResponse from RequestOrResponse
> 
>
> Key: KAFKA-1644
> URL: https://issues.apache.org/jira/browse/KAFKA-1644
> Project: Kafka
>  Issue Type: Bug
>Reporter: Anton Karamanov
>Assignee: Anton Karamanov
> Attachments: 
> 0001-KAFKA-1644-Inherit-FetchResponse-from-RequestOrRespo.patch
>
>
> Unlike all other Kafka API responses {{FetchResponse}} is not a subclass of 
> RequestOrResponse, which requires handling it as a special case while 
> processing responses.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Assigned] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Gwen Shapira (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Gwen Shapira reassigned KAFKA-1664:
---

Assignee: Gwen Shapira

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Ricky Saltzer (JIRA)]

Ricky Saltzer created KAFKA-1664:


 Summary: Kafka does not properly parse multiple ZK nodes with 
non-root chroot
 Key: KAFKA-1664
 URL: https://issues.apache.org/jira/browse/KAFKA-1664
 Project: Kafka
  Issue Type: Bug
  Components: clients
Reporter: Ricky Saltzer
Priority: Blocker


When using a non-root ZK directory for Kafka, if you specify multiple ZK 
servers, Kafka does not seem to properly parse the connection string. 

*Error*
{code}
[root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
 --topic test-topic
[2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
(kafka.consumer.ConsoleConsumer$)
java.lang.IllegalArgumentException: Path length must be > 0
at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
at 
kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
at 
kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
at 
kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
at 
kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
at 
kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
{code}


*Working*
{code}
[root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
{code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Neha Narkhede (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neha Narkhede updated KAFKA-1664:
-
Reviewer: Neha Narkhede

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Neha Narkhede (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neha Narkhede updated KAFKA-1664:
-
Labels: newbie  (was: )

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Gwen Shapira (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155704#comment-14155704
 ] 

Gwen Shapira commented on KAFKA-1664:
-

This happens with other high-level consumer clients. So its not just the 
ConsoleConsumer. 

Topic admin tools don't throw an exception, but don't show topics either.

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Comment Edited] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Gwen Shapira (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155704#comment-14155704
 ] 

Gwen Shapira edited comment on KAFKA-1664 at 10/1/14 10:43 PM:
---

This happens with other high-level consumer clients. So its not just the 
ConsoleConsumer. 

Topic admin tools don't throw an exception, but don't show topics either.

I'm working on a patch.


was (Author: gwenshap):
This happens with other high-level consumer clients. So its not just the 
ConsoleConsumer. 

Topic admin tools don't throw an exception, but don't show topics either.

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-899) LeaderNotAvailableException the first time a new message for a partition is processed.

[Jun Rao (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-899?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155742#comment-14155742
 ] 

Jun Rao commented on KAFKA-899:
---

We started doing that classification in the new java producer. For example, 
there are certain exceptions are of RetriableException. Transient failures like 
leader not available are in that category. Exceptions like MessageTooLarge are 
in a different category. Perhaps you can take a look at that in the new 
producer and see if that makes sense.

> LeaderNotAvailableException the first time a new message for a partition is 
> processed.
> --
>
> Key: KAFKA-899
> URL: https://issues.apache.org/jira/browse/KAFKA-899
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.8.0
>Reporter: Jason Rosenberg
>Assignee: Jun Rao
> Fix For: 0.8.2
>
> Attachments: kafka-899.patch, kafka-899_v2.patch, kafka-899_v3.patch
>
>
> I'm porting some unit tests from 0.7.2 to 0.8.0.  The test does the 
> following, all embedded in the same java process:
> -- spins up a zk instance
> -- spins up a kafka server using a fresh log directory
> -- creates a producer and sends a message
> -- creates a high-level consumer and verifies that it can consume the message
> -- shuts down the consumer
> -- stops the kafka server
> -- stops zk
> The test seems to be working fine now, however, I consistently see the 
> following exceptions (which from poking around the mailing list seem to be 
> expected?).  If these are expected, can we suppress the logging of these 
> exceptions, since it clutters the output of tests, and presumably, clutters 
> the logs of the running server/consumers, during clean startup and 
> shutdown..
> When I call producer.send(), I get:
> 1071 [main] WARN kafka.producer.BrokerPartitionInfo  - Error while fetching 
> metadata  partition 0 leader: nonereplicas:   isr:
> isUnderReplicated: false for topic partition [test-topic,0]: [class 
> kafka.common.LeaderNotAvailableException]
> 1081 [main] WARN kafka.producer.async.DefaultEventHandler  - Failed to 
> collate messages by topic,partition due to
> kafka.common.LeaderNotAvailableException: No leader for any partition
>   at 
> kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$getPartition(DefaultEventHandler.scala:212)
>   at 
> kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:150)
>   at 
> kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:148)
>   at 
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:57)
>   at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:43)
>   at 
> kafka.producer.async.DefaultEventHandler.partitionAndCollate(DefaultEventHandler.scala:148)
>   at 
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:94)
>   at 
> kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:72)
>   at kafka.producer.Producer.send(Producer.scala:74)
>   at kafka.javaapi.producer.Producer.send(Producer.scala:32)
>   at 
> com.squareup.kafka.server.KafkaServerTest.produceAndConsumeMessage(KafkaServerTest.java:98)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
>   at java.lang.reflect.Method.invoke(Method.java:597)
>   at 
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:44)
>   at 
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:15)
>   at 
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:41)
>   at 
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:20)
>   at 
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:28)
>   at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:263)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:69)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:48)
>   at org.junit.runners.ParentRunner$3.run(ParentRunner.java:231)
>   at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:60)
>   at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:229)
>   at org.junit.runners.ParentRunner.access$000(ParentRunner.java:50)
>   at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:22

[jira] [Updated] (KAFKA-1647) Replication offset checkpoints (high water marks) can be lost on hard kills and restarts

[Joel Koshy (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1647?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Joel Koshy updated KAFKA-1647:
--
Assignee: Jiangjie Qin

> Replication offset checkpoints (high water marks) can be lost on hard kills 
> and restarts
> 
>
> Key: KAFKA-1647
> URL: https://issues.apache.org/jira/browse/KAFKA-1647
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.8.2
>Reporter: Joel Koshy
>Assignee: Jiangjie Qin
>Priority: Critical
>  Labels: newbie++
>
> We ran into this scenario recently in a production environment. This can 
> happen when enough brokers in a cluster are taken down. i.e., a rolling 
> bounce done properly should not cause this issue. It can occur if all 
> replicas for any partition are taken down.
> Here is a sample scenario:
> * Cluster of three brokers: b0, b1, b2
> * Two partitions (of some topic) with replication factor two: p0, p1
> * Initial state:
> p0: leader = b0, ISR = {b0, b1}
> p1: leader = b1, ISR = {b0, b1}
> * Do a parallel hard-kill of all brokers
> * Bring up b2, so it is the new controller
> * b2 initializes its controller context and populates its leader/ISR cache 
> (i.e., controllerContext.partitionLeadershipInfo) from zookeeper. The last 
> known leaders are b0 (for p0) and b1 (for p2)
> * Bring up b1
> * The controller's onBrokerStartup procedure initiates a replica state change 
> for all replicas on b1 to become online. As part of this replica state change 
> it gets the last known leader and ISR and sends a LeaderAndIsrRequest to b1 
> (for p1 and p2). This LeaderAndIsr request contains: {{p0: leader=b0; p1: 
> leader=b1;} leaders=b1}. b0 is indicated as the leader of p0 but it is not 
> included in the leaders field because b0 is down.
> * On receiving the LeaderAndIsrRequest, b1's replica manager will 
> successfully make itself (b1) the leader for p1 (and create the local replica 
> object corresponding to p1). It will however abort the become follower 
> transition for p0 because the designated leader b0 is offline. So it will not 
> create the local replica object for p0.
> * It will then start the high water mark checkpoint thread. Since only p1 has 
> a local replica object, only p1's high water mark will be checkpointed to 
> disk. p0's previously written checkpoint  if any will be lost.
> So in summary it seems we should always create the local replica object even 
> if the online transition does not happen.
> Possible symptoms of the above bug could be one or more of the following (we 
> saw 2 and 3):
> # Data loss; yes on a hard-kill data loss is expected, but this can actually 
> cause loss of nearly all data if the broker becomes follower, truncates, and 
> soon after happens to become leader.
> # High IO on brokers that lose their high water mark then subsequently (on a 
> successful become follower transition) truncate their log to zero and start 
> catching up from the beginning.
> # If the offsets topic is affected, then offsets can get reset. This is 
> because during an offset load we don't read past the high water mark. So if a 
> water mark is missing then we don't load anything (even if the offsets are 
> there in the log).



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1644) Inherit FetchResponse from RequestOrResponse

[Jun Rao (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1644?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155903#comment-14155903
 ] 

Jun Rao commented on KAFKA-1644:


Anton,

By letting FetchResponseSend to call FetchResponse.writeTo(), you are no longer 
using the sendfile logic, right?


> Inherit FetchResponse from RequestOrResponse
> 
>
> Key: KAFKA-1644
> URL: https://issues.apache.org/jira/browse/KAFKA-1644
> Project: Kafka
>  Issue Type: Bug
>Reporter: Anton Karamanov
>Assignee: Anton Karamanov
> Attachments: 
> 0001-KAFKA-1644-Inherit-FetchResponse-from-RequestOrRespo.patch
>
>
> Unlike all other Kafka API responses {{FetchResponse}} is not a subclass of 
> RequestOrResponse, which requires handling it as a special case while 
> processing responses.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Jun Rao (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155908#comment-14155908
 ] 

Jun Rao commented on KAFKA-1664:


Actually, namespace should only be included once in a ZK url, not after every 
host:port. So the correct ZK URL should be

 
baelish-001.edh.cloudera.com:2181,baelish-002.edh.cloudera.com:2181,baelish-003.edh.cloudera.com:2181/kafka

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Ricky Saltzer (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14155951#comment-14155951
 ] 

Ricky Saltzer commented on KAFKA-1664:
--

Yes I've tried that as well, same problem.



> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

setting up brokers

[Ranjan Sen]

I have a 5 node cluster with one Jab tracker and another namenode; zookeeper on 
job tracker, namenode and a slave (of the three). 
Can you advise me on what nodes should I use for installing kafka? I am looking 
for at least 2 replications. 
Ranjan

[jira] [Commented] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Jun Rao (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14156022#comment-14156022
 ] 

Jun Rao commented on KAFKA-1664:


So, it seems that namespace doesn't exist. For tools, auto-creating those 
namespaces may not be ideal. Perhaps we just need to provide a more meaningful 
error.

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-1013) Modify existing tools as per the changes in KAFKA-1000

[Mayuresh Gharat (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14156056#comment-14156056
 ] 

Mayuresh Gharat commented on KAFKA-1013:


Updated reviewboard https://reviews.apache.org/r/25944/diff/
 against branch origin/trunk

> Modify existing tools as per the changes in KAFKA-1000
> --
>
> Key: KAFKA-1013
> URL: https://issues.apache.org/jira/browse/KAFKA-1013
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Tejas Patil
>Assignee: Mayuresh Gharat
>Priority: Minor
> Attachments: KAFKA-1013.patch, KAFKA-1013.patch, 
> KAFKA-1013_2014-09-23_10:45:59.patch, KAFKA-1013_2014-09-23_10:48:07.patch, 
> KAFKA-1013_2014-09-26_18:52:09.patch, KAFKA-1013_2014-10-01_21:05:00.patch
>
>
> Modify existing tools as per the changes in KAFKA-1000. AFAIK, the tools 
> below would be affected:
> - ConsumerOffsetChecker
> - ExportZkOffsets
> - ImportZkOffsets
> - UpdateOffsetsInZK



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1013) Modify existing tools as per the changes in KAFKA-1000

[Mayuresh Gharat (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-1013?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Mayuresh Gharat updated KAFKA-1013:
---
Attachment: KAFKA-1013_2014-10-01_21:05:00.patch

> Modify existing tools as per the changes in KAFKA-1000
> --
>
> Key: KAFKA-1013
> URL: https://issues.apache.org/jira/browse/KAFKA-1013
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Tejas Patil
>Assignee: Mayuresh Gharat
>Priority: Minor
> Attachments: KAFKA-1013.patch, KAFKA-1013.patch, 
> KAFKA-1013_2014-09-23_10:45:59.patch, KAFKA-1013_2014-09-23_10:48:07.patch, 
> KAFKA-1013_2014-09-26_18:52:09.patch, KAFKA-1013_2014-10-01_21:05:00.patch
>
>
> Modify existing tools as per the changes in KAFKA-1000. AFAIK, the tools 
> below would be affected:
> - ConsumerOffsetChecker
> - ExportZkOffsets
> - ImportZkOffsets
> - UpdateOffsetsInZK



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: Review Request 25944: Patch for KAFKA-1013

[Mayuresh Gharat]

---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/25944/
---

(Updated Oct. 2, 2014, 4:05 a.m.)


Review request for kafka.


Bugs: KAFKA-1013
https://issues.apache.org/jira/browse/KAFKA-1013


Repository: kafka


Description (updated)
---

Made changes to client, ExportOffsets,ImportOffsets and test cases


Diffs (updated)
-

  core/src/main/scala/kafka/client/ClientUtils.scala 
ebba87f0566684c796c26cb76c64b4640a5ccfde 
  core/src/main/scala/kafka/consumer/SimpleConsumer.scala 
d349a3000feb9ccd57d1f3cb163548d5bf432186 
  core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala 
fbc680fde21b02f11285a4f4b442987356abd17b 
  core/src/main/scala/kafka/javaapi/consumer/SimpleConsumer.scala 
0ab0195dc9f66f407061d0fac2549bd6291e41ff 
  core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala 
d1e7c434e77859d746b8dc68dd5d5a3740425e79 
  core/src/main/scala/kafka/tools/ExportOffsets.scala PRE-CREATION 
  core/src/main/scala/kafka/tools/ExportZkOffsets.scala 
4d051bc2db12f0e15aa6a3289abeb9dd25d373d2 
  core/src/main/scala/kafka/tools/ImportOffsets.scala PRE-CREATION 
  core/src/main/scala/kafka/tools/ImportZkOffsets.scala 
abe09721b13f71320510fd1a01c1917470450c6e 
  core/src/main/scala/kafka/tools/OffsetClient.scala PRE-CREATION 
  core/src/main/scala/kafka/tools/OffsetClientConfig.scala PRE-CREATION 
  core/src/main/scala/kafka/utils/Utils.scala 
29d5a17d4a03cfd3f3cdd2994cbd783a4be2732e 
  core/src/test/scala/other/kafka/TestOffsetManager.scala 
41f334d48897b3027ed54c58bbf4811487d3b191 
  core/src/test/scala/unit/kafka/server/OffsetCommitTest.scala 
2d9325045ac1ac2d7531161b32c98c847125cbf0 

Diff: https://reviews.apache.org/r/25944/diff/


Testing
---


Thanks,

Mayuresh Gharat

[jira] [Commented] (KAFKA-1664) Kafka does not properly parse multiple ZK nodes with non-root chroot

[Neha Narkhede (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-1664?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14156057#comment-14156057
 ] 

Neha Narkhede commented on KAFKA-1664:
--

+1 on providing a more meaningful error.

> Kafka does not properly parse multiple ZK nodes with non-root chroot
> 
>
> Key: KAFKA-1664
> URL: https://issues.apache.org/jira/browse/KAFKA-1664
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Ricky Saltzer
>Assignee: Gwen Shapira
>Priority: Blocker
>  Labels: newbie
>
> When using a non-root ZK directory for Kafka, if you specify multiple ZK 
> servers, Kafka does not seem to properly parse the connection string. 
> *Error*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka,baelish-002.edh.cloudera.com:2181/kafka,baelish-003.edh.cloudera.com:2181/kafka
>  --topic test-topic
> [2014-10-01 15:31:04,629] ERROR Error processing message, stopping consumer:  
> (kafka.consumer.ConsoleConsumer$)
> java.lang.IllegalArgumentException: Path length must be > 0
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:48)
>   at org.apache.zookeeper.common.PathUtils.validatePath(PathUtils.java:35)
>   at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:766)
>   at org.I0Itec.zkclient.ZkConnection.create(ZkConnection.java:87)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:308)
>   at org.I0Itec.zkclient.ZkClient$1.call(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:675)
>   at org.I0Itec.zkclient.ZkClient.create(ZkClient.java:304)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:213)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at org.I0Itec.zkclient.ZkClient.createPersistent(ZkClient.java:223)
>   at kafka.utils.ZkUtils$.createParentPath(ZkUtils.scala:245)
>   at kafka.utils.ZkUtils$.createEphemeralPath(ZkUtils.scala:256)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflict(ZkUtils.scala:268)
>   at 
> kafka.utils.ZkUtils$.createEphemeralPathExpectConflictHandleZKBug(ZkUtils.scala:306)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$registerConsumerInZK(ZookeeperConsumerConnector.scala:226)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:755)
>   at 
> kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:145)
>   at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:196)
>   at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)
> {code}
> *Working*
> {code}
> [root@hodor-001 bin]# ./kafka-console-consumer.sh --zookeeper 
> baelish-001.edh.cloudera.com:2181/kafka --topic test-topic
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: Review Request 25886: KAFKA-1555: provide strong consistency with reasonable availability

[Jun Rao]

---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/25886/#review55187
---


Thanks for the patch. A few comments below.


clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java


"all" is more meaningful than -1. So, we can keep it that way. Under the 
cover, we turn "all" to -1 in the producer request.



core/src/main/scala/kafka/cluster/Partition.scala


Hmm, this code looks a bit weird. It's weird in that a topic level config 
only takes effect under certain producer side config. My feeling is that if we 
make min.isr a topic level config, then it should work in the same way 
independent of the producer side config. So, perhaps it's better to just check 
inSyncSize < minIsr, irrespective of the ack. What do you think?



core/src/main/scala/kafka/log/LogConfig.scala


To be consistent with other per topic configs, we need to define it in 
KafkaConfig as a global setting.


- Jun Rao


On Oct. 1, 2014, 1:19 a.m., Gwen Shapira wrote:
> 
> ---
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/25886/
> ---
> 
> (Updated Oct. 1, 2014, 1:19 a.m.)
> 
> 
> Review request for kafka.
> 
> 
> Repository: kafka
> 
> 
> Description
> ---
> 
> KAFKA-1555: provide strong consistency with reasonable availability
> 
> 
> Diffs
> -
> 
>   clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java 
> f9de4af 
>   clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java addc906 
>   
> clients/src/main/java/org/apache/kafka/common/errors/NotEnoughReplicasAfterAppendException.java
>  PRE-CREATION 
>   
> clients/src/main/java/org/apache/kafka/common/errors/NotEnoughReplicasException.java
>  PRE-CREATION 
>   clients/src/main/java/org/apache/kafka/common/protocol/Errors.java d434f42 
>   core/src/main/scala/kafka/cluster/Partition.scala ff106b4 
>   core/src/main/scala/kafka/common/ErrorMapping.scala 3fae791 
>   
> core/src/main/scala/kafka/common/NotEnoughReplicasAfterAppendException.scala 
> PRE-CREATION 
>   core/src/main/scala/kafka/common/NotEnoughReplicasException.scala 
> PRE-CREATION 
>   core/src/main/scala/kafka/log/LogConfig.scala 5746ad4 
>   core/src/main/scala/kafka/producer/SyncProducerConfig.scala 69b2d0c 
>   core/src/main/scala/kafka/server/KafkaApis.scala c584b55 
>   core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala 
> 39f777b 
>   core/src/test/scala/unit/kafka/producer/ProducerTest.scala dd71d81 
>   core/src/test/scala/unit/kafka/producer/SyncProducerTest.scala 24deea0 
>   core/src/test/scala/unit/kafka/utils/TestUtils.scala 2dbdd3c 
> 
> Diff: https://reviews.apache.org/r/25886/diff/
> 
> 
> Testing
> ---
> 
> With 3 broker cluster, created 3 topics each with 1 partition and 3 replicas, 
> with 1,3 and 4 min.insync.replicas.
> * min.insync.replicas=1 behaved normally (all writes succeeded as long as a 
> broker was up)
> * min.insync.replicas=3 returned NotEnoughReplicas when required.acks=-1 and 
> one broker was down
> * min.insync.replicas=4 returned NotEnoughReplicas when required.acks=-1
> 
> See notes about retry behavior in the JIRA.
> 
> 
> Thanks,
> 
> Gwen Shapira
> 
>

Re: Two open issues on Kafka security

[Jay Kreps]

Hey Jarek,

I agree with the importance of separating authentication and
authorization. The question is what concept of identity is sufficient
to pass through to the authorization layer? Just a "user name"? Or
perhaps you also need the ip the request originated from? Whatever
these would be it would be nice to enumerate them so the authz portion
can be written in a way that ignores the authn part.

So if no one else proposes anything different maybe we can just say
user name + ip?

With respect to hierarchy, it would be nice to have topic hierarchies
but we don't have them now so seems overkill to try to think them
through wrt security now, right?

-Jay



On Wed, Oct 1, 2014 at 1:13 PM, Jarek Jarcec Cecho  wrote:
> I’m following the security proposal wiki page [1] and this discussion and I 
> would like to jump in with few points if I might :)  Let me start by saying 
> that I like the material and the discussion here, good work!
>
> I was part of the team who originally designed and worked on Sentry and I 
> wanted to share few to see how it will resonate with people.  My first and 
> probably biggest point would be to separate authorization and authentication 
> as two separate systems. I believe that Jao has already stressed that in the 
> email thread, but I wanted to reiterate on that point. In my experience users 
> don’t care that much about how the user has been authenticated if they trust 
> that mechanism, what they care more about is that the authorization model is 
> consistent and behaves the same way. E.g. if I configured that user jarcec 
> can write into topic “logs”, he should be able to do that no matter where the 
> connection came from - whether he has been authorized from Kerberos as he is 
> directly exploring the data from his computer, he is authorized through 
> delegation token because he is running map reduce jobs calculating statistics 
> or he is  authorized through SSL certificated because … (well I’m missing 
> good example here, but you’re probably following my point).
>
> I’ve also noticed that we are planning to have no hierarchy in the authz 
> object model per the wiki [1] with the reasoning that Kafka do not supports 
> topic hierarchy. I see that point, but at the same time it got me thinking - 
> are we sure that Kafka will never have hierarchic topics? Seems as a nice 
> feature that might be usable for some use cases and something that we might 
> want to add in the future. But regardless of that I would suggest to 
> introduce a hierarchy anyway, even though if it would be just two levels. In 
> sentry (for Hive) we’ve introduced concept of “Service” where all the 
> databases are children of the service. In Kafka I would imagine that we would 
> have “service” and “topics” as the children. Having this is much easier to 
> model general privileges where you need to grant access to all topics - you 
> will just grant access to the entire service and all topics will get 
> “inherited”.
>
> I’m wondering what are other people thoughts?
>
> Jarcec
>
> Links:
> 1: https://cwiki.apache.org/confluence/display/KAFKA/Security
>
> On Oct 1, 2014, at 9:44 AM, Joe Stein  wrote:
>
>> Hi Jonathan,
>>
>> "Hadoop delegation tokens to enable MapReduce, Samza, or other frameworks
>> running in the Hadoop environment to access Kafka"
>> https://cwiki.apache.org/confluence/display/KAFKA/Security is on the list,
>> yup!
>>
>> /***
>> Joe Stein
>> Founder, Principal Consultant
>> Big Data Open Source Security LLC
>> http://www.stealth.ly
>> Twitter: @allthingshadoop 
>> /
>>
>> On Wed, Oct 1, 2014 at 12:35 PM, Jonathan Creasy 
>> wrote:
>>
>>> This is not nearly as deep as the discussion so far, but I did want to
>>> throw this idea out there to make sure we¹ve thought about it.
>>>
>>> The Kafka project should make sure that when deployed alongside a Hadoop
>>> cluster from any major distributions that it can tie seamlessly into the
>>> authentication and authorization used within that cluster. For example,
>>> Apache Sentry.
>>>
>>> This may present additional difficulties that means a decision is made to
>>> not do that or alternatively the Kerberos authentication and the
>>> authorization schemes we are already working on may be sufficient.
>>>
>>> I¹m not sure that anything I¹ve read so far in this discussion actually
>>> poses a problem, but I¹m an Ops guy and being able to more easily
>>> integrate more things, makes my life better. :)
>>>
>>> -Jonathan
>>>
>>> On 9/30/14, 11:26 PM, "Joe Stein"  wrote:
>>>
 inline

 On Tue, Sep 30, 2014 at 11:58 PM, Jay Kreps  wrote:

> Hey Joe,
>
> For (1) what are you thinking for the PermissionManager api?
>
> The way I see it, the first question we have to answer is whether it
> is possible to make authentication and authorization independent. What
> I mean by that is whether 

Re: Two open issues on Kafka security

[Jay Kreps]

Hey Michael,

WRT question 2, I think for SASL you do need the mechanism information
but what I was talking about was the challenge/response byte[] that is
sent back and forth from the client to the server. My understanding is
that SASL gives you an api for the client and server to use to produce
these byte[]'s but doesn't actually specify any way of exchanging them
(that is protocol specific). I could be wrong here since my knowledge
of this stuff is pretty weak. But according to my understanding you
must be imagining some protocol for exchanging challenge/response
information. This protocol would have to be clearly documented for
client implementors. What is that protocol?

-Jay

On Wed, Oct 1, 2014 at 2:36 PM, Michael Herstine
 wrote:
> Regarding question #1, I’m not sure I follow you, Joe: you’re proposing (I
> think) that the API take a byte[], but what will be in that array? A
> serialized certificate if the client authenticated via SSL and the
> principal name (perhaps normalized) if the client authenticated via
> Kerberos?
>
> Regarding question #2, I think I was unclear in the meeting yesterday: I
> was proposing a separate port for each authentication method (including
> none). That is, if a client wants no authentication, then they would
> connect to port N on the broker. If they wanted to talk over SSL, then
> they connect to port N+1 (say). Kerberos: N+2. This would remove the need
> for a new request, since the authentication type would be implicit in the
> port on which the client connected (and it was my understanding that it
> was desirable to not introduce any new messages).
>
> Perhaps the confusion comes from the fact, correctly pointed out by Jay,
> that when you want to use SASL on a single port, there does of course need
> to be a way for the incoming client to signal which mechanism it wants to
> use, and that’s out of scope of the SASL spec. I didn’t see there being a
> desire to add new SASL mechanisms going forward, but perhaps I was
> incorrect?
>
> In any event, I’d like to suggest we keep the “open” or “no auth” port
> separate, both to make it easy for admins to force the use of security (by
> shutting down that port) and to avoid downgrade attacks (where an attacker
> intercepts the opening packet from a client requesting security & alters
> it to request none).
>
> I’ll update the Wiki with my notes from yesterday’s meeting this afternoon.
>
> Thanks,
>
> On 10/1/14, 9:35 AM, "Jonathan Creasy"  wrote:
>
>>This is not nearly as deep as the discussion so far, but I did want to
>>throw this idea out there to make sure we¹ve thought about it.
>>
>>The Kafka project should make sure that when deployed alongside a Hadoop
>>cluster from any major distributions that it can tie seamlessly into the
>>authentication and authorization used within that cluster. For example,
>>Apache Sentry.
>>
>>This may present additional difficulties that means a decision is made to
>>not do that or alternatively the Kerberos authentication and the
>>authorization schemes we are already working on may be sufficient.
>>
>>I¹m not sure that anything I¹ve read so far in this discussion actually
>>poses a problem, but I¹m an Ops guy and being able to more easily
>>integrate more things, makes my life better. :)
>>
>>-Jonathan
>>
>>On 9/30/14, 11:26 PM, "Joe Stein"  wrote:
>>
>>>inline
>>>
>>>On Tue, Sep 30, 2014 at 11:58 PM, Jay Kreps  wrote:
>>>
 Hey Joe,

 For (1) what are you thinking for the PermissionManager api?

 The way I see it, the first question we have to answer is whether it
 is possible to make authentication and authorization independent. What
 I mean by that is whether I can write an authorization library that
 will work the same whether you authenticate with ssl or kerberos.
>>>
>>>
>>>To me that is a requirement. We can't tie them together.  We have to
>>>provide the ability for authorization to work regardless of the
>>>authentication.  One *VERY* important use case is level of trust in
>>>authentication from the authorization perpsective.  e.g. I authorize
>>>"identity" based on the how you authenticated Alice is able to view
>>>topic X if Alice authenticated over kerberos.  Bob isn't allowed to view
>>>topic X no matter what. Alice can authenticate over not kerberos (uses
>>>cases for that) and in that case Alice wouldn't see topic X.  A concrete
>>>use case for this with Kafka would be a third party bank consuming data
>>>to
>>>a broker.  The service provider would have some kerberos local auth for
>>>that bank to-do back up that would also have access to other topics
>>>related
>>>to that banks data the bank itself over SSL wants a stream of events
>>>(some specific topic) and that banks identity only sees that topic.  It
>>>is
>>>important to not confuse identity, authentication and authorization.
>>>
>>>
 If
 so then we need to pick some subset of identity information that we
 can extract from both and have this constitute the id

Re: Two open issues on Kafka security

[Jay Kreps]

Here is the client side in ZK:
https://svn.apache.org/repos/asf/zookeeper/trunk/src/java/main/org/apache/zookeeper/client/ZooKeeperSaslClient.java

Note how they have a special Zookeeper request API that is used to
send the SASL bytes (e.g. see ZooKeeperSaslClient.sendSaslPacket).

This API follows the same protocol and rpc mechanism all their other
request/response types follow but it just has a simple byte[] entry
for the SASL token in both the request and response.

-Jay

On Wed, Oct 1, 2014 at 9:46 PM, Jay Kreps  wrote:
> Hey Michael,
>
> WRT question 2, I think for SASL you do need the mechanism information
> but what I was talking about was the challenge/response byte[] that is
> sent back and forth from the client to the server. My understanding is
> that SASL gives you an api for the client and server to use to produce
> these byte[]'s but doesn't actually specify any way of exchanging them
> (that is protocol specific). I could be wrong here since my knowledge
> of this stuff is pretty weak. But according to my understanding you
> must be imagining some protocol for exchanging challenge/response
> information. This protocol would have to be clearly documented for
> client implementors. What is that protocol?
>
> -Jay
>
> On Wed, Oct 1, 2014 at 2:36 PM, Michael Herstine
>  wrote:
>> Regarding question #1, I’m not sure I follow you, Joe: you’re proposing (I
>> think) that the API take a byte[], but what will be in that array? A
>> serialized certificate if the client authenticated via SSL and the
>> principal name (perhaps normalized) if the client authenticated via
>> Kerberos?
>>
>> Regarding question #2, I think I was unclear in the meeting yesterday: I
>> was proposing a separate port for each authentication method (including
>> none). That is, if a client wants no authentication, then they would
>> connect to port N on the broker. If they wanted to talk over SSL, then
>> they connect to port N+1 (say). Kerberos: N+2. This would remove the need
>> for a new request, since the authentication type would be implicit in the
>> port on which the client connected (and it was my understanding that it
>> was desirable to not introduce any new messages).
>>
>> Perhaps the confusion comes from the fact, correctly pointed out by Jay,
>> that when you want to use SASL on a single port, there does of course need
>> to be a way for the incoming client to signal which mechanism it wants to
>> use, and that’s out of scope of the SASL spec. I didn’t see there being a
>> desire to add new SASL mechanisms going forward, but perhaps I was
>> incorrect?
>>
>> In any event, I’d like to suggest we keep the “open” or “no auth” port
>> separate, both to make it easy for admins to force the use of security (by
>> shutting down that port) and to avoid downgrade attacks (where an attacker
>> intercepts the opening packet from a client requesting security & alters
>> it to request none).
>>
>> I’ll update the Wiki with my notes from yesterday’s meeting this afternoon.
>>
>> Thanks,
>>
>> On 10/1/14, 9:35 AM, "Jonathan Creasy"  wrote:
>>
>>>This is not nearly as deep as the discussion so far, but I did want to
>>>throw this idea out there to make sure we¹ve thought about it.
>>>
>>>The Kafka project should make sure that when deployed alongside a Hadoop
>>>cluster from any major distributions that it can tie seamlessly into the
>>>authentication and authorization used within that cluster. For example,
>>>Apache Sentry.
>>>
>>>This may present additional difficulties that means a decision is made to
>>>not do that or alternatively the Kerberos authentication and the
>>>authorization schemes we are already working on may be sufficient.
>>>
>>>I¹m not sure that anything I¹ve read so far in this discussion actually
>>>poses a problem, but I¹m an Ops guy and being able to more easily
>>>integrate more things, makes my life better. :)
>>>
>>>-Jonathan
>>>
>>>On 9/30/14, 11:26 PM, "Joe Stein"  wrote:
>>>
inline

On Tue, Sep 30, 2014 at 11:58 PM, Jay Kreps  wrote:

> Hey Joe,
>
> For (1) what are you thinking for the PermissionManager api?
>
> The way I see it, the first question we have to answer is whether it
> is possible to make authentication and authorization independent. What
> I mean by that is whether I can write an authorization library that
> will work the same whether you authenticate with ssl or kerberos.


To me that is a requirement. We can't tie them together.  We have to
provide the ability for authorization to work regardless of the
authentication.  One *VERY* important use case is level of trust in
authentication from the authorization perpsective.  e.g. I authorize
"identity" based on the how you authenticated Alice is able to view
topic X if Alice authenticated over kerberos.  Bob isn't allowed to view
topic X no matter what. Alice can authenticate over not kerberos (uses
cases for that) and in that case Alice wouldn't se