[jira] [Commented] (KAFKA-2686) unsubscribe() call leaves KafkaConsumer in invalid state for manual topic-partition assignment

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2686:
---

GitHub user guozhangwang opened a pull request:

https://github.com/apache/kafka/pull/352

KAFKA-2686: Reset needsPartitionAssignment in SubscriptionState.assign()



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/guozhangwang/kafka K2686

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/352.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #352


commit ecc94205c0731ecf25737307aac2ffda20fc1a14
Author: Guozhang Wang 
Date:   2015-10-22T19:05:50Z

v1




> unsubscribe() call leaves KafkaConsumer in invalid state for manual 
> topic-partition assignment
> --
>
> Key: KAFKA-2686
> URL: https://issues.apache.org/jira/browse/KAFKA-2686
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: The Data Lorax
>Assignee: Guozhang Wang
>
> The bellow code snippet demonstrated the problem.
> Basically, the unsubscribe() call leaves the KafkaConsumer in a state that 
> means poll() will always return empty record sets, even if new 
> topic-partitions have been assigned that have messages pending.  This is 
> because unsubscribe() sets SubscriptionState.needsPartitionAssignment to 
> true, and assign() does not clear this flag. The only thing that clears this 
> flag is when the consumer handles the response from a JoinGroup request.
> {code}
> final KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
> ConsumerRecords records = consumer.poll(100);// <- Works, 
> returning records
> consumer.unsubscribe();   // Puts consumer into invalid state.
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 2)));
> records = consumer.poll(100);// <- Always returns empty record set.
> {code}



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

Re: [VOTE] KIP-38: ZooKeeper authentication

[Edward Ribeiro]

+1 (non-binding)

On Thu, Oct 22, 2015 at 3:46 PM, Neha Narkhede  wrote:

> +1 (binding)
>
> On Wed, Oct 21, 2015 at 7:24 PM, Todd Palino  wrote:
>
> > +1 (non-binding)
> >
> > On Wed, Oct 21, 2015 at 6:53 PM, Jiangjie Qin  >
> > wrote:
> >
> > > +1 (non-binding)
> > >
> > > On Wed, Oct 21, 2015 at 3:40 PM, Joel Koshy 
> wrote:
> > >
> > > > +1 binding
> > > >
> > > > On Wed, Oct 21, 2015 at 8:17 AM, Flavio Junqueira 
> > > wrote:
> > > >
> > > > > Thanks everyone for the feedback so far. At this point, I'd like to
> > > start
> > > > > a vote for KIP-38.
> > > > >
> > > > > Summary: Add support for ZooKeeper authentication
> > > > > KIP page:
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38%3A+ZooKeeper+Authentication
> > > > > <
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38:+ZooKeeper+Authentication
> > > > > >
> > > > >
> > > > > Thanks,
> > > > > -Flavio
> > > >
> > >
> >
>
>
>
> --
> Thanks,
> Neha
>

Re: [DISCUSS] KIP-38: ZooKeeper Authentication

[Mayuresh Gharat]

This might have been explained before. I had a question :
In the KIP it says :
"One ZooKeeper setting of interest on the server side is
zookeeper.allowSaslFailedClients. If this is false, then clients trying to
authenticate with an incorrect configuration will have their connections
dropped. Otherwise, such clients will be able to connect successfully, but
will not have the right credentials set. Setting it to false prevents
clients with an incorrect configuration from making progress.ZooKeeper also
allows users to disable authentication on the client side even in the
presence of a JAAS login file with the property zookeeper.sasl.client.
Setting it to false disables client authentication."


*1) So if I set "zookeeper.sasl.client" to false and
"zookeeper.allowSaslFailedClients" to true, will the clients be
authenticated? *
*2) Also what is meant by clients with incorrect configuration?*

Thanks,

Mayuresh

On Thu, Oct 22, 2015 at 8:11 AM, Flavio Junqueira  wrote:

> Ok, thanks for spotting it.
>
> -Flavio
>
> > On 22 Oct 2015, at 05:54, Jun Rao  wrote:
> >
> > It seems that in the secure -> unsecure plan, step 3 needs to be done
> > before step 2.
> >
> > Thanks,
> >
> > Jun
> >
> > On Wed, Oct 21, 2015 at 3:59 PM, Flavio Junqueira 
> wrote:
> >
> >> Ok, thanks for the feedback, Todd. I have updated the KIP with some of
> the
> >> points discussed here. There is more to add based on these last
> comments,
> >> though.
> >>
> >> -Flavio
> >>
> >>> On 21 Oct 2015, at 23:43, Todd Palino  wrote:
> >>>
> >>> On Wed, Oct 21, 2015 at 3:38 PM, Flavio Junqueira  >> > wrote:
> >>>
> 
> > On 21 Oct 2015, at 21:54, Todd Palino  wrote:
> >
> > Thanks for the clarification on that, Jun. Obviously, we haven't been
>  doing
> > much with ZK authentication around here yet. There is still a small
>  concern
> > there, mostly in that you should not share credentials any more than
> is
> > necessary, which would argue for being able to use a different ACL
> than
>  the
> > default. I don't really like the idea of having to use the exact same
> > credentials for executing the admin tools as we do for running the
>  brokers.
> > Given that we don't need to share the credentials with all
> consumers, I
> > think we can work around it.
> >
> 
>  Let me add that a feature to separate the sub-trees of users sharing
> an
>  ensemble is chroot.
> 
>  On different credentials for admin tools, this sounds doable by
> setting
>  the ACLs of znodes. For example, there could be an admin id and a
> broker
>  id, both with the ability of changing znodes, but different
> credentials.
>  Would something like that work for you?
> 
> >>>
> >>> It would be a nice option to have, as the credentials can be protected
> >>> differently. I would consider this a nice to have, and not an
> "absolutely
> >>> must have" feature at this point.
> >>>
> >>>
>  This does bring up another good question, however. What will be the
>  process
> > for having to rotate the credentials? That is, if the credentials are
> > compromised and need to be changed, how can that be accomplished with
> >> the
> > cluster online. I'm guessing some combination of using skipAcl on the
> > Zookeeper ensemble and config changes to the brokers will be
> required,
>  but
> > this is an important enough operation that we should make sure it's
> > reasonable to perform and that it is documented.
> 
>  Right now there is no kafka support in the plan for this. But this is
>  doable directly through the zk api. Would it be sufficient to write
> down
>  how to perform such an operation via the zk api or do we need a tool
> to
> >> do
>  it?
> 
> >>>
> >>> I think as long as there is a documented procedure for how to do it,
> that
> >>> will be good enough. It's mostly about making sure that we can, and
> that
> >> we
> >>> don't put something in place that would require downtime to a cluster
> in
> >>> order to change credentials. We can always develop a tool later if it
> is
> >> a
> >>> requested item.
> >>>
> >>> Thanks!
> >>>
> >>> -Todd
> >>>
> >>>
> >>>
> 
>  -Flavio
> 
> >
> >
> > On Wed, Oct 21, 2015 at 1:23 PM, Jun Rao  wrote:
> >
> >> Parth,
> >>
> >> For 2), in your approach, the broker/controller will then always
> have
>  the
> >> overhead of resetting the ACL on startup after zookeeper.set.acl is
> >> set
>  to
> >> true. The benefit of using a separate migration tool is that you
> paid
>  the
> >> cost only once during upgrade. It is an extra step during the
> upgrade.
> >> However, given the other things that you need to do to upgrade to
> >> 0.9.0
> >> (e.g. two rounds of rolling upgrades on 

Re: [DISCUSS] KIP-32 - Add CreateTime and LogAppendTime to Kafka message

[Jay Kreps]

Hey Kartik,

Yes, I agree exactly with your characterization.

The question is "what is the meaning of retention?" It could mean either:
1. "Retain data that is no more than 7 days old"
2. "Retain data for 7 days from when you get it"

I don't know if either is actually a clear winner.

Each is intuitive and easily expressible:
1. "We have the last 7 days of data in Kafka"
2. "You have 7 days to get your data from Kafka from whenever it arrives"

Each has corner cases:
1. May lead to retaining data for too little time in the bootstrap case
2. May lead to over-retention in the bootstrap case

Which failure is worse probably depends on who you ask:
1. May lead to not retaining data long enough for a consumer to get it,
which the consumer would say is very bad
2. May lead to running out of disk space, which ops would say is very bad

-Jay




On Tue, Oct 20, 2015 at 12:02 AM, Kartik Paramasivam <
kparamasi...@linkedin.com.invalid> wrote:

> Joel or Becket will probably respond back in more detail.. but here are my
> 2c.
>
> From the standpoint of LinkedIN, the suggested proposal works.. in essence
> max.appenddelay can be used to turn "creationTime" into "logAppendTime".
>this does mean that at LinkedIn we won't be able to use "creationTime"..
> however that might also be fine because we anyways use the timeStamp that
> is set inside the avro payload.
>
> Keeping LI aside though, it looks like there are two distinct possible
> goals.
> 1. The broker will retain messages for x days after a message shows up at
> the broker.   This behavior would super deterministic and would never
> change depending on the contents of the message or anything else.
>
> 2. The client is in "partial" control of how long a message stays in the
> broker based on the creationTime stamped by the client.
>
> Although (2) could be a feature in some scenarios..but in many scenarios it
> can be pretty unintuitive and be perceived as an anti-feature.  For e.g say
> a mobile client buffered up some messages because the device was offline
> (maybe in a plane).. and then sent the message after say 23 hours on a
> plane.  The message shows up in a Kafka topic with 24 hour retention.. and
> now the message gets deleted in 1 hour.
>
> Kartik
>
>
> On Tue, Oct 13, 2015 at 12:32 AM, Jay Kreps  wrote:
>
> > Here's my basic take:
> > - I agree it would be nice to have a notion of time baked in if it were
> > done right
> > - All the proposals so far seem pretty complex--I think they might make
> > things worse rather than better overall
> > - I think adding 2x8 byte timestamps to the message is probably a
> > non-starter from a size perspective
> > - Even if it isn't in the message, having two notions of time that
> control
> > different things is a bit confusing
> > - The mechanics of basing retention etc on log append time when that's
> not
> > in the log seem complicated
> >
> > To that end here is a possible 4th option. Let me know what you think.
> >
> > The basic idea is that the message creation time is closest to what the
> > user actually cares about but is dangerous if set wrong. So rather than
> > substitute another notion of time, let's try to ensure the correctness of
> > message creation time by preventing arbitrarily bad message creation
> times.
> >
> > First, let's see if we can agree that log append time is not something
> > anyone really cares about but rather an implementation detail. The
> > timestamp that matters to the user is when the message occurred (the
> > creation time). The log append time is basically just an approximation to
> > this on the assumption that the message creation and the message receive
> on
> > the server occur pretty close together and the reason to prefer .
> >
> > But as these values diverge the issue starts to become apparent. Say you
> > set the retention to one week and then mirror data from a topic
> containing
> > two years of retention. Your intention is clearly to keep the last week,
> > but because the mirroring is appending right now you will keep two years.
> >
> > The reason we are liking log append time is because we are (justifiably)
> > concerned that in certain situations the creation time may not be
> > trustworthy. This same problem exists on the servers but there are fewer
> > servers and they just run the kafka code so it is less of an issue.
> >
> > There are two possible ways to handle this:
> >
> >1. Just tell people to add size based retention. I think this is not
> >entirely unreasonable, we're basically saying we retain data based on
> > the
> >timestamp you give us in the data. If you give us bad data we will
> > retain
> >it for a bad amount of time. If you want to ensure we don't retain
> "too
> >much" data, define "too much" by setting a time-based retention
> setting.
> >This is not entirely unreasonable but kind of suffers from a "one bad
> >apple" problem in a very large environment.
> >2. Prevent bad timestamps. 

Build failed in Jenkins: kafka-trunk-jdk7 #711

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-2678; partition level lag metrics can be negative

--
[...truncated 2826 lines...]

kafka.consumer.TopicFilterTest > testWhitelists PASSED

kafka.consumer.TopicFilterTest > 
testWildcardTopicCountGetTopicCountMapEscapeJson PASSED

kafka.consumer.TopicFilterTest > testBlacklists PASSED

kafka.consumer.PartitionAssignorTest > testRoundRobinPartitionAssignor PASSED

kafka.consumer.PartitionAssignorTest > testRangePartitionAssignor PASSED

kafka.consumer.MetricsTest > testMetricsReporterAfterDeletingTopic PASSED

kafka.consumer.MetricsTest > testMetricsLeak PASSED

kafka.common.TopicTest > testInvalidTopicNames PASSED

kafka.common.TopicTest > testTopicHasCollision PASSED

kafka.common.TopicTest > testTopicHasCollisionChars PASSED

kafka.common.ConfigTest > testInvalidGroupIds PASSED

kafka.common.ConfigTest > testInvalidClientIds PASSED

kafka.common.ZkNodeChangeNotificationListenerTest > testProcessNotification 
PASSED

kafka.zk.ZKEphemeralTest > testZkWatchedEphemeralRecursive PASSED

kafka.zk.ZKEphemeralTest > testOverlappingSessions PASSED

kafka.zk.ZKEphemeralTest > testEphemeralNodeCleanup PASSED

kafka.zk.ZKEphemeralTest > testZkWatchedEphemeral PASSED

kafka.zk.ZKEphemeralTest > testSameSession PASSED

kafka.zk.ZKPathTest > testCreatePersistentSequentialThrowsException PASSED

kafka.zk.ZKPathTest > testCreatePersistentSequentialExists PASSED

kafka.zk.ZKPathTest > testCreateEphemeralPathExists PASSED

kafka.zk.ZKPathTest > testCreatePersistentPath PASSED

kafka.zk.ZKPathTest > testMakeSurePersistsPathExistsThrowsException PASSED

kafka.zk.ZKPathTest > testCreateEphemeralPathThrowsException PASSED

kafka.zk.ZKPathTest > testCreatePersistentPathThrowsException PASSED

kafka.zk.ZKPathTest > testMakeSurePersistsPathExists PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testAllowAllAccess PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testNoAclFound PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testAclManagementAPIs PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testWildCardAcls PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testTopicAcl PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testSuperUserHasAccess PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testDenyTakesPrecedence PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testNoAclFoundOverride PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testLoadCache PASSED

kafka.server.SaslSslReplicaFetchTest > testReplicaFetcherThread PASSED

kafka.server.ReplicaManagerTest > testHighWaterMarkDirectoryMapping PASSED

kafka.server.ReplicaManagerTest > testIllegalRequiredAcks PASSED

kafka.server.ReplicaManagerTest > testHighwaterMarkRelativeDirectoryMapping 
PASSED

kafka.server.IsrExpirationTest > testIsrExpirationForSlowFollowers PASSED

kafka.server.IsrExpirationTest > testIsrExpirationForStuckFollowers PASSED

kafka.server.IsrExpirationTest > testIsrExpirationIfNoFetchRequestMade PASSED

kafka.server.ClientQuotaManagerTest > testQuotaViolation PASSED

kafka.server.ClientQuotaManagerTest > testQuotaParsing PASSED

kafka.server.SimpleFetchTest > testReadFromLog PASSED

kafka.server.ServerGenerateBrokerIdTest > testAutoGenerateBrokerId PASSED

kafka.server.ServerGenerateBrokerIdTest > testMultipleLogDirsMetaProps PASSED

kafka.server.ServerGenerateBrokerIdTest > testUserConfigAndGeneratedBrokerId 
PASSED

kafka.server.ServerGenerateBrokerIdTest > 
testConsistentBrokerIdFromUserConfigAndMetaProps PASSED

kafka.server.ThrottledResponseExpirationTest > testThrottledRequest PASSED

kafka.server.ThrottledResponseExpirationTest > testExpire PASSED

kafka.server.SslReplicaFetchTest > testReplicaFetcherThread PASSED

kafka.server.ServerShutdownTest > testCleanShutdownAfterFailedStartup PASSED

kafka.server.ServerShutdownTest > testConsecutiveShutdown PASSED

kafka.server.ServerShutdownTest > testCleanShutdown PASSED

kafka.server.ServerShutdownTest > testCleanShutdownWithDeleteTopicEnabled PASSED

kafka.server.KafkaConfigTest > testAdvertiseConfigured PASSED

kafka.server.KafkaConfigTest > testLogRetentionTimeHoursProvided PASSED

kafka.server.KafkaConfigTest > testLogRollTimeBothMsAndHoursProvided PASSED

kafka.server.KafkaConfigTest > testLogRetentionValid PASSED

kafka.server.KafkaConfigTest > testSpecificProperties PASSED

kafka.server.KafkaConfigTest > testDefaultCompressionType PASSED

kafka.server.KafkaConfigTest > testDuplicateListeners PASSED

kafka.server.KafkaConfigTest > testLogRetentionUnlimited PASSED

kafka.server.KafkaConfigTest > testLogRetentionTimeMsProvided PASSED

kafka.server.KafkaConfigTest > testLogRollTimeNoConfigProvided PASSED

kafka.server.KafkaConfigTest > testAdvertiseDefaults PASSED

kafka.server.KafkaConfigTest > testBadListenerProtocol PASSED

kafka.server.KafkaConfigTest > testListenerDefaults PASSED

kafka.server.KafkaConfigTest > 

[jira] [Updated] (KAFKA-2686) unsubscribe() call leaves KafkaConsumer in invalid state for manual topic-partition assignment

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-2686:
-
Reviewer: Jason Gustafson

> unsubscribe() call leaves KafkaConsumer in invalid state for manual 
> topic-partition assignment
> --
>
> Key: KAFKA-2686
> URL: https://issues.apache.org/jira/browse/KAFKA-2686
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: The Data Lorax
>Assignee: Guozhang Wang
>
> The bellow code snippet demonstrated the problem.
> Basically, the unsubscribe() call leaves the KafkaConsumer in a state that 
> means poll() will always return empty record sets, even if new 
> topic-partitions have been assigned that have messages pending.  This is 
> because unsubscribe() sets SubscriptionState.needsPartitionAssignment to 
> true, and assign() does not clear this flag. The only thing that clears this 
> flag is when the consumer handles the response from a JoinGroup request.
> {code}
> final KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
> ConsumerRecords records = consumer.poll(100);// <- Works, 
> returning records
> consumer.unsubscribe();   // Puts consumer into invalid state.
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 2)));
> records = consumer.poll(100);// <- Always returns empty record set.
> {code}



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

[GitHub] kafka pull request: KAFKA-2686: Reset needsPartitionAssignment in ...

[guozhangwang]

GitHub user guozhangwang opened a pull request:

https://github.com/apache/kafka/pull/352

KAFKA-2686: Reset needsPartitionAssignment in SubscriptionState.assign()



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/guozhangwang/kafka K2686

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/352.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #352


commit ecc94205c0731ecf25737307aac2ffda20fc1a14
Author: Guozhang Wang 
Date:   2015-10-22T19:05:50Z

v1




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Question about compressed messages...

[Robert Thille]

I’m working on a Twisted Python Kafka client and I was wondering what the ‘key’ 
on a gzip’d block of messages “means”.  

That is, if the client has a batch of messages to send, with a mix of keys, 
would it be a bug to batch them together and gzip into a single message?  Or is 
the key on the outer “message” ignored/should be set to Null?

Thanks,

Robert

—
Robert P. Thille | Senior Software Engineer, Blue Planet
rthi...@ciena.com  | 1383 N. McDowell Blvd. Suite 300 
| Petaluma, CA 94954
Direct +1.707.735.2300 | Mobile +1.707.861.0042 

Re: [DISCUSS] KIP-38: ZooKeeper Authentication

[Flavio Junqueira]

Hi Mayuresh, 

See comments below, please:

> On 22 Oct 2015, at 18:17, Mayuresh Gharat  wrote:
> 
> This might have been explained before. I had a question :
> In the KIP it says :
> "One ZooKeeper setting of interest on the server side is
> zookeeper.allowSaslFailedClients. If this is false, then clients trying to
> authenticate with an incorrect configuration will have their connections
> dropped. Otherwise, such clients will be able to connect successfully, but
> will not have the right credentials set. Setting it to false prevents
> clients with an incorrect configuration from making progress.ZooKeeper also
> allows users to disable authentication on the client side even in the
> presence of a JAAS login file with the property zookeeper.sasl.client.
> Setting it to false disables client authentication."
> 
> 
> *1) So if I set "zookeeper.sasl.client" to false and
> "zookeeper.allowSaslFailedClients" to true, will the clients be
> authenticated? *

"zookeeper.sasl.client" is a config parameter on the client side, while 
"allowSaslFailedClients" is a configuration parameter for servers. If you set 
"zookeeper.sasl.client" to false, then the client won't be authenticated 
independent of the value of "allowSaslFailedClients" because you'll be 
disabling the SASL client code. 

> *2) Also what is meant by clients with incorrect configuration?*
> 

Say you make a mistake in your JAAS login file. If "allowSaslFailedClients" is 
set to true, then your client would still connect (not authenticated) silently, 
and you may prefer to spot that problem right up front by having the server 
reject the connection. 

-Flavio

> 
> On Thu, Oct 22, 2015 at 8:11 AM, Flavio Junqueira  wrote:
> 
>> Ok, thanks for spotting it.
>> 
>> -Flavio
>> 
>>> On 22 Oct 2015, at 05:54, Jun Rao  wrote:
>>> 
>>> It seems that in the secure -> unsecure plan, step 3 needs to be done
>>> before step 2.
>>> 
>>> Thanks,
>>> 
>>> Jun
>>> 
>>> On Wed, Oct 21, 2015 at 3:59 PM, Flavio Junqueira 
>> wrote:
>>> 
 Ok, thanks for the feedback, Todd. I have updated the KIP with some of
>> the
 points discussed here. There is more to add based on these last
>> comments,
 though.
 
 -Flavio
 
> On 21 Oct 2015, at 23:43, Todd Palino  wrote:
> 
> On Wed, Oct 21, 2015 at 3:38 PM, Flavio Junqueira > wrote:
> 
>> 
>>> On 21 Oct 2015, at 21:54, Todd Palino  wrote:
>>> 
>>> Thanks for the clarification on that, Jun. Obviously, we haven't been
>> doing
>>> much with ZK authentication around here yet. There is still a small
>> concern
>>> there, mostly in that you should not share credentials any more than
>> is
>>> necessary, which would argue for being able to use a different ACL
>> than
>> the
>>> default. I don't really like the idea of having to use the exact same
>>> credentials for executing the admin tools as we do for running the
>> brokers.
>>> Given that we don't need to share the credentials with all
>> consumers, I
>>> think we can work around it.
>>> 
>> 
>> Let me add that a feature to separate the sub-trees of users sharing
>> an
>> ensemble is chroot.
>> 
>> On different credentials for admin tools, this sounds doable by
>> setting
>> the ACLs of znodes. For example, there could be an admin id and a
>> broker
>> id, both with the ability of changing znodes, but different
>> credentials.
>> Would something like that work for you?
>> 
> 
> It would be a nice option to have, as the credentials can be protected
> differently. I would consider this a nice to have, and not an
>> "absolutely
> must have" feature at this point.
> 
> 
>> This does bring up another good question, however. What will be the
>> process
>>> for having to rotate the credentials? That is, if the credentials are
>>> compromised and need to be changed, how can that be accomplished with
 the
>>> cluster online. I'm guessing some combination of using skipAcl on the
>>> Zookeeper ensemble and config changes to the brokers will be
>> required,
>> but
>>> this is an important enough operation that we should make sure it's
>>> reasonable to perform and that it is documented.
>> 
>> Right now there is no kafka support in the plan for this. But this is
>> doable directly through the zk api. Would it be sufficient to write
>> down
>> how to perform such an operation via the zk api or do we need a tool
>> to
 do
>> it?
>> 
> 
> I think as long as there is a documented procedure for how to do it,
>> that
> will be good enough. It's mostly about making sure that we can, and
>> that
 we
> don't put something in place that would require downtime to a cluster
>> in
> order to 

[jira] [Assigned] (KAFKA-2686) unsubscribe() call leaves KafkaConsumer in invalid state for manual topic-partition assignment

[Guozhang Wang (JIRA)]

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

Guozhang Wang reassigned KAFKA-2686:


Assignee: Guozhang Wang  (was: Neha Narkhede)

> unsubscribe() call leaves KafkaConsumer in invalid state for manual 
> topic-partition assignment
> --
>
> Key: KAFKA-2686
> URL: https://issues.apache.org/jira/browse/KAFKA-2686
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: The Data Lorax
>Assignee: Guozhang Wang
>
> The bellow code snippet demonstrated the problem.
> Basically, the unsubscribe() call leaves the KafkaConsumer in a state that 
> means poll() will always return empty record sets, even if new 
> topic-partitions have been assigned that have messages pending.  This is 
> because unsubscribe() sets SubscriptionState.needsPartitionAssignment to 
> true, and assign() does not clear this flag. The only thing that clears this 
> flag is when the consumer handles the response from a JoinGroup request.
> {code}
> final KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
> ConsumerRecords records = consumer.poll(100);// <- Works, 
> returning records
> consumer.unsubscribe();   // Puts consumer into invalid state.
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 2)));
> records = consumer.poll(100);// <- Always returns empty record set.
> {code}



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

[GitHub] kafka pull request: KAFKA-2652: integrate new group protocol into ...

[ymatsuda]

GitHub user ymatsuda opened a pull request:

https://github.com/apache/kafka/pull/353

KAFKA-2652: integrate new group protocol into partition grouping

@guozhangwang 

* added ```PartitionGrouper``` (abstract class)
 * This class is responsible for grouping partitions to form tasks.
 * Users may implement this class for custom grouping.
* added ```DefaultPartitionGrouper```
 * our default implementation of ```PartitionGrouper```
* added ```KafkaStreamingPartitionAssignor```
 * We always use this as ```PartitionAssignor``` of stream consumers.
 * Actual grouping is delegated to ```PartitionGrouper```.
* ```TopologyBuilder```
 * added ```topicGroups()```
  * This returns groups of related topics according to the topology
 * added ```copartitionSources(sourceNodes...)```
  * This is used by DSL layer. It asserts the specified source nodes must 
be copartitioned.
 * added ```copartitionGroups()``` which returns groups of copartitioned 
topics
* KStream layer
 * keep track of source nodes to determine copartition sources when steams 
are joined
 * source nodes are set to null when partitioning property is not preserved 
(ex. ```map()```, ```transform()```), and this indicates the stream is no 
longer joinable


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/ymatsuda/kafka grouping

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/353.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #353


commit 708718c1be23fad25fa6206f665cbb619c1b5097
Author: Yasuhiro Matsuda 
Date:   2015-10-19T19:38:06Z

partition grouping

commit d2bae046b5509022e2821a2c5eb08853d228e791
Author: Yasuhiro Matsuda 
Date:   2015-10-19T20:19:54Z

wip

commit 86fa8110b23ee1992fbd19daa08c63a4b427448e
Author: Yasuhiro Matsuda 
Date:   2015-10-20T20:01:37Z

long task id

commit 4f4f9ac642ebe0eae33a5c8464309106e9239f2e
Author: Yasuhiro Matsuda 
Date:   2015-10-20T20:03:15Z

Merge branch 'trunk' of github.com:apache/kafka into grouping

commit e4ecf39b9ab0b0f4c915a4f43cfe771b1de69f7f
Author: Yasuhiro Matsuda 
Date:   2015-10-21T19:33:05Z

joinability

commit 37d72a691173a8fe878ac3d99e8973e72f5675c6
Author: Yasuhiro Matsuda 
Date:   2015-10-21T19:33:48Z

Merge branch 'trunk' of github.com:apache/kafka into grouping

commit f68723bab83c3a3f1c15872f4f24bc932df8198f
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:21:31Z

partition assignor

commit 457cf270222139eae89750781d09abaa07120932
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:21:40Z

Merge branch 'trunk' of github.com:apache/kafka into grouping

commit 13f3ad703960581229d511287f27345c567b5d3e
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:34:52Z

complete undoing long taskid

commit 98f3bcc1896fd159ccbbd37fc65b1d9d6f568bb9
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:45:38Z

fix a test




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-2652) Incorporate the new consumer protocol with partition-group interface

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2652:
---

GitHub user ymatsuda opened a pull request:

https://github.com/apache/kafka/pull/353

KAFKA-2652: integrate new group protocol into partition grouping

@guozhangwang 

* added ```PartitionGrouper``` (abstract class)
 * This class is responsible for grouping partitions to form tasks.
 * Users may implement this class for custom grouping.
* added ```DefaultPartitionGrouper```
 * our default implementation of ```PartitionGrouper```
* added ```KafkaStreamingPartitionAssignor```
 * We always use this as ```PartitionAssignor``` of stream consumers.
 * Actual grouping is delegated to ```PartitionGrouper```.
* ```TopologyBuilder```
 * added ```topicGroups()```
  * This returns groups of related topics according to the topology
 * added ```copartitionSources(sourceNodes...)```
  * This is used by DSL layer. It asserts the specified source nodes must 
be copartitioned.
 * added ```copartitionGroups()``` which returns groups of copartitioned 
topics
* KStream layer
 * keep track of source nodes to determine copartition sources when steams 
are joined
 * source nodes are set to null when partitioning property is not preserved 
(ex. ```map()```, ```transform()```), and this indicates the stream is no 
longer joinable


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/ymatsuda/kafka grouping

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/353.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #353


commit 708718c1be23fad25fa6206f665cbb619c1b5097
Author: Yasuhiro Matsuda 
Date:   2015-10-19T19:38:06Z

partition grouping

commit d2bae046b5509022e2821a2c5eb08853d228e791
Author: Yasuhiro Matsuda 
Date:   2015-10-19T20:19:54Z

wip

commit 86fa8110b23ee1992fbd19daa08c63a4b427448e
Author: Yasuhiro Matsuda 
Date:   2015-10-20T20:01:37Z

long task id

commit 4f4f9ac642ebe0eae33a5c8464309106e9239f2e
Author: Yasuhiro Matsuda 
Date:   2015-10-20T20:03:15Z

Merge branch 'trunk' of github.com:apache/kafka into grouping

commit e4ecf39b9ab0b0f4c915a4f43cfe771b1de69f7f
Author: Yasuhiro Matsuda 
Date:   2015-10-21T19:33:05Z

joinability

commit 37d72a691173a8fe878ac3d99e8973e72f5675c6
Author: Yasuhiro Matsuda 
Date:   2015-10-21T19:33:48Z

Merge branch 'trunk' of github.com:apache/kafka into grouping

commit f68723bab83c3a3f1c15872f4f24bc932df8198f
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:21:31Z

partition assignor

commit 457cf270222139eae89750781d09abaa07120932
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:21:40Z

Merge branch 'trunk' of github.com:apache/kafka into grouping

commit 13f3ad703960581229d511287f27345c567b5d3e
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:34:52Z

complete undoing long taskid

commit 98f3bcc1896fd159ccbbd37fc65b1d9d6f568bb9
Author: Yasuhiro Matsuda 
Date:   2015-10-22T18:45:38Z

fix a test




> Incorporate the new consumer protocol with partition-group interface
> 
>
> Key: KAFKA-2652
> URL: https://issues.apache.org/jira/browse/KAFKA-2652
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Guozhang Wang
>Assignee: Yasuhiro Matsuda
> Fix For: 0.9.0.1
>
>
> After KAFKA-2464 is checked in, we need to incorporate the new protocol along 
> with a partition-group interface.
> The first step maybe a couple of pre-defined partitioning scheme that can be 
> chosen by user from some configs.



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

Build failed in Jenkins: kafka-trunk-jdk8 #50

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-2678; partition level lag metrics can be negative

[wangguoz] MINOR: Restore `SslConsumerTest` which was accidentally deleted in

--
[...truncated 5329 lines...]

kafka.coordinator.MemberMetadataTest > testVoteRaisesOnNoSupportedProtocols 
PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[0] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[1] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[2] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[3] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[4] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[5] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[6] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[7] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[8] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[9] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[10] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[11] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[12] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[13] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[14] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[15] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[16] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[17] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[18] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[19] PASSED

kafka.log.OffsetMapTest > testClear PASSED

kafka.log.OffsetMapTest > testBasicValidation PASSED

kafka.log.LogManagerTest > testCleanupSegmentsToMaintainSize PASSED

kafka.log.LogManagerTest > testRecoveryDirectoryMappingWithRelativeDirectory 
PASSED

kafka.log.LogManagerTest > testGetNonExistentLog PASSED

kafka.log.LogManagerTest > testTwoLogManagersUsingSameDirFails PASSED

kafka.log.LogManagerTest > testLeastLoadedAssignment PASSED

kafka.log.LogManagerTest > testCleanupExpiredSegments PASSED

kafka.log.LogManagerTest > testCheckpointRecoveryPoints PASSED

kafka.log.LogManagerTest > testTimeBasedFlush PASSED

kafka.log.LogManagerTest > testCreateLog PASSED

kafka.log.LogManagerTest > testRecoveryDirectoryMappingWithTrailingSlash PASSED

kafka.log.LogSegmentTest > testRecoveryWithCorruptMessage PASSED

kafka.log.LogSegmentTest > testRecoveryFixesCorruptIndex PASSED

kafka.log.LogSegmentTest > testReadFromGap PASSED

kafka.log.LogSegmentTest > testTruncate PASSED

kafka.log.LogSegmentTest > testReadBeforeFirstOffset PASSED

kafka.log.LogSegmentTest > testCreateWithInitFileSizeAppendMessage PASSED

kafka.log.LogSegmentTest > testChangeFileSuffixes PASSED

kafka.log.LogSegmentTest > testMaxOffset PASSED

kafka.log.LogSegmentTest > testNextOffsetCalculation PASSED

kafka.log.LogSegmentTest > testReadOnEmptySegment PASSED

kafka.log.LogSegmentTest > testReadAfterLast PASSED

kafka.log.LogSegmentTest > testCreateWithInitFileSizeClearShutdown PASSED

kafka.log.LogSegmentTest > testTruncateFull PASSED

kafka.log.FileMessageSetTest > testTruncate PASSED

kafka.log.FileMessageSetTest > testIterationOverPartialAndTruncation PASSED

kafka.log.FileMessageSetTest > testRead PASSED

kafka.log.FileMessageSetTest > testFileSize PASSED

kafka.log.FileMessageSetTest > testIteratorWithLimits PASSED

kafka.log.FileMessageSetTest > testPreallocateTrue PASSED

kafka.log.FileMessageSetTest > testIteratorIsConsistent PASSED

kafka.log.FileMessageSetTest > testIterationDoesntChangePosition PASSED

kafka.log.FileMessageSetTest > testWrittenEqualsRead PASSED

kafka.log.FileMessageSetTest > testWriteTo PASSED

kafka.log.FileMessageSetTest > testPreallocateFalse PASSED

kafka.log.FileMessageSetTest > testPreallocateClearShutdown PASSED

kafka.log.FileMessageSetTest > testSearch PASSED

kafka.log.FileMessageSetTest > testSizeInBytes PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[0] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[1] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[2] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[3] PASSED

kafka.log.LogTest > testParseTopicPartitionNameForMissingTopic PASSED

kafka.log.LogTest > testIndexRebuild PASSED

kafka.log.LogTest > testLogRolls PASSED

kafka.log.LogTest > testMessageSizeCheck PASSED

kafka.log.LogTest > testAsyncDelete PASSED

kafka.log.LogTest > testReadOutOfRange PASSED

kafka.log.LogTest > testReadAtLogGap PASSED

kafka.log.LogTest > testTimeBasedLogRoll PASSED

kafka.log.LogTest > testLoadEmptyLog PASSED

kafka.log.LogTest > testMessageSetSizeCheck PASSED

kafka.log.LogTest > testIndexResizingAtTruncation PASSED

kafka.log.LogTest > 

Re: [VOTE] KIP-38: ZooKeeper authentication

[Neha Narkhede]

+1 (binding)

On Wed, Oct 21, 2015 at 7:24 PM, Todd Palino  wrote:

> +1 (non-binding)
>
> On Wed, Oct 21, 2015 at 6:53 PM, Jiangjie Qin 
> wrote:
>
> > +1 (non-binding)
> >
> > On Wed, Oct 21, 2015 at 3:40 PM, Joel Koshy  wrote:
> >
> > > +1 binding
> > >
> > > On Wed, Oct 21, 2015 at 8:17 AM, Flavio Junqueira 
> > wrote:
> > >
> > > > Thanks everyone for the feedback so far. At this point, I'd like to
> > start
> > > > a vote for KIP-38.
> > > >
> > > > Summary: Add support for ZooKeeper authentication
> > > > KIP page:
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38%3A+ZooKeeper+Authentication
> > > > <
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38:+ZooKeeper+Authentication
> > > > >
> > > >
> > > > Thanks,
> > > > -Flavio
> > >
> >
>



-- 
Thanks,
Neha

Jenkins build is back to normal : kafka-trunk-jdk7 #712

[Apache Jenkins Server]

See 

[jira] [Assigned] (KAFKA-2685) "alter topic" on non-existent topic exits without error

[Edward Ribeiro (JIRA)]

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

Edward Ribeiro reassigned KAFKA-2685:
-

Assignee: Edward Ribeiro

> "alter topic" on non-existent topic exits without error
> ---
>
> Key: KAFKA-2685
> URL: https://issues.apache.org/jira/browse/KAFKA-2685
> Project: Kafka
>  Issue Type: Bug
>Reporter: Gwen Shapira
>Assignee: Edward Ribeiro
>
> When running:
> kafka-topics --zookeeper localhost:2181 --alter --topic test --config 
> unclean.leader.election.enable=false
> and topic "test" does not exist, the command simply return with no error 
> message.
> We expect to see an error when trying to modify non-existing topics, so user 
> will have a chance to catch and correct typos.



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

Re: [DISCUSS] KIP-32 - Add CreateTime and LogAppendTime to Kafka message

[Jay Kreps]

Hey Joel,

I think we're mostly in agreement. I don't know about you but I'm not
head-over-heals in love with any of these proposals, including mine--time
is just pretty complicated and icky.

With respect to using log append time here was basically how I felt:
1. I think in a vacuum basing offset lookup on append time is worse than
using create time. If I'm looking for messages that happened at 12:00, I
don't really care when they were mirrored to a particular server. This kind
of time based lookup is definitely an important aspect.
2. I think in a vacuum basing retention off create time is not as good as
basing it off log append time, though there are drawbacks to both as I
outlined in the other email.
3. I think having multiple notions of time isn't ideal. Offset is already
one notion of time which has super nice properties. I get adding another
but adding two more isn't great.
4. I don't particularly care about log rolling as I think the primary
motivation for it is removed by the time-based index (as I described in the
other email).
5. My concern with prop log append time was primarily that there was kind
of a lot of mechanism and adding that stuff to the fetch request just
seemed a bit complicated and a somewhat hacky. My concern is more an unease
that this would complicate life going forward and have more issues than we
would anticipate rather than any concrete problem.

As you point out, and as I tried to summarize, for bootstrapping the
tradeoff is between over-retention and under-retention of data.

-Jay


On Tue, Oct 20, 2015 at 12:09 AM, Joel Koshy  wrote:

>  I’m in favor of adding the create-time in the message (although some would
> argue even that should really be an application-level header), but I don’t
> think it should be mutable after it leaves the client and I think we should
> avoid having the server use that for any server-side indexing. The
> max.append.delay config helps, but I wouldn’t be surprised if it ends up
> becoming a very confusing configuration.
>
> I also agree with being thrifty with headers and that having both
> create-time and log-append-time in the header is overkill which is what the
> third option addresses. I’m not fully convinced that the implementation
> details of basing retention on log append time (i.e., the third option) are
> terribly complicated. Can you describe your concerns with the earlier
> approach? While it is true that using log-append-time to drive retention
> won’t handle the bootstrap case, it is a clear approach in that it makes no
> promises about that scenario - i.e., retention/rolling/offset lookup will
> all be based on arrival time at the server and not message creation
> time. There are use-cases which don’t care about log append time, but for
> those use-cases the create time (if we add it) will be available in each
> message. It’s just that retention/rolling will be driven off log append
> time.
>
> Becket has already brought out some scenarios where the usage of
> create-time in combination with max.append.delay may be ambiguous and
> unintuitive. Here are some others: say if we use the
> create-time-driven-index for retention; if a new segment gets created with
> time t1 and a message arrives out of create-time order (t0 < t1). Then the
> second message will be held hostage until t1 + retention so retention is
> violated. I actually may be completely unclear on how retention should work
> with create-time-driven-indexes and max.append.delay. Say we set
> max.append.delay to something high (or infinity). Wouldn’t the user have to
> set retention appropriately as well? Otherwise really old (by create-time)
> messages that are say from a bootstrapping source would just get purged
> shortly after arrival. So if max.append.delay is infinity it seems the
> right retention setting is also infinity. Can you clarify how retention
> should work if driven off an index that is built from create-time?
>
> Also wrt the max.append.delay - allowing the server to override it would
> render the create-time field pretty much untrustworthy right? The most
> intuitive policy I can think of for create time is to make it immutable
> after it has been set at the sender. Otherwise we would need to add some
> dirty flag or a generation field in order to be able to distinguish between
> messages with the true create-time and the ones that are not - but that
> seems to be a hack that suggests it should be immutable in the first place.
>
> Thanks,
>
> Joel
>
> On Tue, Oct 20, 2015 at 12:02 AM, Kartik Paramasivam <
> kparamasi...@linkedin.com.invalid> wrote:
>
> > Joel or Becket will probably respond back in more detail.. but here are
> my
> > 2c.
> >
> > From the standpoint of LinkedIN, the suggested proposal works.. in
> essence
> > max.appenddelay can be used to turn "creationTime" into "logAppendTime".
> >this does mean that at LinkedIn we won't be able to use
> "creationTime"..
> > however that might also be fine because we anyways 

Re: [DISCUSS] KIP-32 - Add CreateTime and LogAppendTime to Kafka message

[Jay Kreps]

Hey Becket,

Let me see if I can address your concerns:

1. Let's say we have two source clusters that are mirrored to the same
> target cluster. For some reason one of the mirror maker from a cluster dies
> and after fix the issue we want to resume mirroring. In this case it is
> possible that when the mirror maker resumes mirroring, the timestamp of the
> messages have already gone beyond the acceptable timestamp range on broker.
> In order to let those messages go through, we have to bump up the
> *max.append.delay
> *for all the topics on the target broker. This could be painful.


Actually what I was suggesting was different. Here is my observation:
clusters/topics directly produced to by applications have a valid assertion
that log append time and create time are similar (let's call these
"unbuffered"); other cluster/topic such as those that receive data from a
database, a log file, or another kafka cluster don't have that assertion,
for these "buffered" clusters data can be arbitrarily late. This means any
use of log append time on these buffered clusters is not very meaningful,
and create time and log append time "should" be similar on unbuffered
clusters so you can probably use either.

Using log append time on buffered clusters actually results in bad things.
If you request the offset for a given time you get don't end up getting
data for that time but rather data that showed up at that time. If you try
to retain 7 days of data it may mostly work but any kind of bootstrapping
will result in retaining much more (potentially the whole database
contents!).

So what I am suggesting in terms of the use of the max.append.delay is that
unbuffered clusters would have this set and buffered clusters would not. In
other words, in LI terminology, tracking and metrics clusters would have
this enforced, aggregate and replica clusters wouldn't.

So you DO have the issue of potentially maintaining more data than you need
to on aggregate clusters if your mirroring skews, but you DON'T need to
tweak the setting as you described.

2. Let's say in the above scenario we let the messages in, at that point
> some log segments in the target cluster might have a wide range of
> timestamps, like Guozhang mentioned the log rolling could be tricky because
> the first time index entry does not necessarily have the smallest timestamp
> of all the messages in the log segment. Instead, it is the largest
> timestamp ever seen. We have to scan the entire log to find the message
> with smallest offset to see if we should roll.


I think there are two uses for time-based log rolling:
1. Making the offset lookup by timestamp work
2. Ensuring we don't retain data indefinitely if it is supposed to get
purged after 7 days

But think about these two use cases. (1) is totally obviated by the
time=>offset index we are adding which yields much more granular offset
lookups. (2) Is actually totally broken if you switch to append time,
right? If you want to be sure for security/privacy reasons you only retain
7 days of data then if the log append and create time diverge you actually
violate this requirement.

I think 95% of people care about (1) which is solved in the proposal and
(2) is actually broken today as well as in both proposals.

3. Theoretically it is possible that an older log segment contains
> timestamps that are older than all the messages in a newer log segment. It
> would be weird that we are supposed to delete the newer log segment before
> we delete the older log segment.


The index timestamps would always be a lower bound (i.e. the maximum at
that time) so I don't think that is possible.

 4. In bootstrap case, if we reload the data to a Kafka cluster, we have to
> make sure we configure the topic correctly before we load the data.
> Otherwise the message might either be rejected because the timestamp is too
> old, or it might be deleted immediately because the retention time has
> reached.


See (1).

-Jay

On Tue, Oct 13, 2015 at 7:30 PM, Jiangjie Qin 
wrote:

> Hey Jay and Guozhang,
>
> Thanks a lot for the reply. So if I understand correctly, Jay's proposal
> is:
>
> 1. Let client stamp the message create time.
> 2. Broker build index based on client-stamped message create time.
> 3. Broker only takes message whose create time is withing current time
> plus/minus T (T is a configuration *max.append.delay*, could be topic level
> configuration), if the timestamp is out of this range, broker rejects the
> message.
> 4. Because the create time of messages can be out of order, when broker
> builds the time based index it only provides the guarantee that if a
> consumer starts consuming from the offset returned by searching by
> timestamp t, they will not miss any message created after t, but might see
> some messages created before t.
>
> To build the time based index, every time when a broker needs to insert a
> new time index entry, the entry would be {Largest_Timestamp_Ever_Seen ->
> 

[jira] [Work started] (KAFKA-2675) SASL/Kerberos follow-up

[Ismael Juma (JIRA)]

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

Work on KAFKA-2675 started by Ismael Juma.
--
> SASL/Kerberos follow-up
> ---
>
> Key: KAFKA-2675
> URL: https://issues.apache.org/jira/browse/KAFKA-2675
> Project: Kafka
>  Issue Type: Sub-task
>  Components: security
>Reporter: Ismael Juma
>Assignee: Ismael Juma
> Fix For: 0.9.0.0
>
>
> This is a follow-up to KAFKA-1686. 
> 1. Decide on `serviceName` configuration: do we want to keep it in two places?
> 2. auth.to.local config name is a bit opaque, is there a better one?
> 3. Implement or remove SASL_KAFKA_SERVER_REALM config
> 4. Consider making Login's thread a daemon thread
> 5. Write test that shows authentication failure due to invalid user
> 6. Write test that shows authentication failure due to wrong password
> 7. Write test that shows authentication failure due ticket expiring



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

Re: [VOTE] KIP-38: ZooKeeper authentication

[Jun Rao]

+1

Thanks,

Jun

On Wed, Oct 21, 2015 at 8:17 AM, Flavio Junqueira  wrote:

> Thanks everyone for the feedback so far. At this point, I'd like to start
> a vote for KIP-38.
>
> Summary: Add support for ZooKeeper authentication
> KIP page:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38%3A+ZooKeeper+Authentication
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38:+ZooKeeper+Authentication
> >
>
> Thanks,
> -Flavio

Re: Java high level consumer providing duplicate messages when auto commit is off

[Cliff Rhyne]

We did some more testing with logging turned on (I figured out why it
wasn't working).  We tried increasing the JVM memory capacity on our test
server (it's lower than in production) and increasing the zookeeper
timeouts.  Neither changed the results.  With trace logging enabled, we saw
that we were getting rebalances even though there is only one high level
consumer running (there previously was a simple consumer that was told to
disconnect, but that consumer only checked the offsets and never consumed
data).

- Is there possibly a race condition where the simple consumer has a hold
on a partition and shutdown is called before starting a high level consumer
but shutdown is done asynchronously?
- What are the various things that can cause a consumer rebalance other
than adding / removing high level consumers?

Thanks,
Cliff

On Wed, Oct 21, 2015 at 4:20 PM, Cliff Rhyne  wrote:

> Hi Kris,
>
> Thanks for the tip.  I'm going to investigate this further.  I checked and
> we have fairly short zk timeouts and run with a smaller memory allocation
> on the two environments we encounter this issue.  I'll let you all know
> what I find.
>
> I saw this ticket https://issues.apache.org/jira/browse/KAFKA-2049 that
> seems to be related to the problem (but would only inform that an issue
> occurred).  Are there any other open issues that could be worked on to
> improve Kafka's handling of this situation?
>
> Thanks,
> Cliff
>
> On Wed, Oct 21, 2015 at 2:53 PM, Kris K  wrote:
>
>> Hi Cliff,
>>
>> One other case I observed in my environment is - when there were gc pauses
>> on one of our high level consumer in the group.
>>
>> Thanks,
>> Kris
>>
>> On Wed, Oct 21, 2015 at 10:12 AM, Cliff Rhyne  wrote:
>>
>> > Hi James,
>> >
>> > There are two scenarios we run:
>> >
>> > 1. Multiple partitions with one consumer per partition.  This rarely has
>> > starting/stopping of consumers, so the pool is very static.  There is a
>> > configured consumer timeout, which is causing the
>> ConsumerTimeoutException
>> > to get thrown prior to the test starting.  We handle this exception and
>> > then resume consuming.
>> > 2. Single partition with one consumer.  This consumer is started by a
>> > triggered condition (number of messages pending to be processed in the
>> > kafka topic or a schedule).  The consumer is stopped after processing is
>> > completed.
>> >
>> > In both cases, based on my understanding there shouldn't be a rebalance
>> as
>> > either a) all consumers are running or b) there's only one consumer /
>> > partition.  Also, the same consumer group is used by all consumers in
>> > scenario 1 and 2.  Is there a good way to investigate whether rebalances
>> > are occurring?
>> >
>> > Thanks,
>> > Cliff
>> >
>> > On Wed, Oct 21, 2015 at 11:37 AM, James Cheng  wrote:
>> >
>> > > Do you have multiple consumers in a consumer group?
>> > >
>> > > I think that when a new consumer joins the consumer group, that the
>> > > existing consumers will stop consuming during the group rebalance, and
>> > then
>> > > when they start consuming again, that they will consume from the last
>> > > committed offset.
>> > >
>> > > You should get more verification on this, tho. I might be remembering
>> > > wrong.
>> > >
>> > > -James
>> > >
>> > > > On Oct 21, 2015, at 8:40 AM, Cliff Rhyne  wrote:
>> > > >
>> > > > Hi,
>> > > >
>> > > > My team and I are looking into a problem where the Java high level
>> > > consumer
>> > > > provides duplicate messages if we turn auto commit off (using
>> version
>> > > > 0.8.2.1 of the server and Java client).  The expected sequence of
>> > events
>> > > > are:
>> > > >
>> > > > 1. Start high-level consumer and initialize a KafkaStream to get a
>> > > > ConsumerIterator
>> > > > 2. Consume n items (could be 10,000, could be 1,000,000) from the
>> > > iterator
>> > > > 3. Commit the new offsets
>> > > >
>> > > > What we are seeing is that during step 2, some number of the n
>> messages
>> > > are
>> > > > getting returned by the iterator in duplicate (in some cases, we've
>> > seen
>> > > > n*5 messages consumed).  The problem appears to go away if we turn
>> on
>> > > auto
>> > > > commit (and committing offsets to kafka helped too), but auto commit
>> > > causes
>> > > > conflicts with our offset rollback logic.  The issue seems to happen
>> > more
>> > > > when we are in our test environment on a lower-cost cloud provider.
>> > > >
>> > > > Diving into the Java and Scala classes including the
>> ConsumerIterator,
>> > > it's
>> > > > not obvious what event causes a duplicate offset to be requested or
>> > > > returned (there's even a loop that is supposed to exclude duplicate
>> > > > messages in this class).  I tried turning on trace logging but my
>> log4j
>> > > > config isn't getting the Kafka client logs to write out.
>> > > >
>> > > > Does anyone have suggestions of where to look or how to enable
>> logging?

Re: [VOTE] KIP-38: ZooKeeper authentication

[Ashish Singh]

+1 (non-binding)

On Thu, Oct 22, 2015 at 12:10 PM, Edward Ribeiro 
wrote:

> +1 (non-binding)
>
> On Thu, Oct 22, 2015 at 3:46 PM, Neha Narkhede  wrote:
>
> > +1 (binding)
> >
> > On Wed, Oct 21, 2015 at 7:24 PM, Todd Palino  wrote:
> >
> > > +1 (non-binding)
> > >
> > > On Wed, Oct 21, 2015 at 6:53 PM, Jiangjie Qin
>  > >
> > > wrote:
> > >
> > > > +1 (non-binding)
> > > >
> > > > On Wed, Oct 21, 2015 at 3:40 PM, Joel Koshy 
> > wrote:
> > > >
> > > > > +1 binding
> > > > >
> > > > > On Wed, Oct 21, 2015 at 8:17 AM, Flavio Junqueira 
> > > > wrote:
> > > > >
> > > > > > Thanks everyone for the feedback so far. At this point, I'd like
> to
> > > > start
> > > > > > a vote for KIP-38.
> > > > > >
> > > > > > Summary: Add support for ZooKeeper authentication
> > > > > > KIP page:
> > > > > >
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38%3A+ZooKeeper+Authentication
> > > > > > <
> > > > > >
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-38:+ZooKeeper+Authentication
> > > > > > >
> > > > > >
> > > > > > Thanks,
> > > > > > -Flavio
> > > > >
> > > >
> > >
> >
> >
> >
> > --
> > Thanks,
> > Neha
> >
>



-- 

Regards,
Ashish

[jira] [Commented] (KAFKA-2675) SASL/Kerberos follow-up

[Jun Rao (JIRA)]

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

Jun Rao commented on KAFKA-2675:


For #4, making the login thread a daemon thread is also fine.

> SASL/Kerberos follow-up
> ---
>
> Key: KAFKA-2675
> URL: https://issues.apache.org/jira/browse/KAFKA-2675
> Project: Kafka
>  Issue Type: Sub-task
>  Components: security
>Reporter: Ismael Juma
>Assignee: Ismael Juma
> Fix For: 0.9.0.0
>
>
> This is a follow-up to KAFKA-1686. 
> 1. Decide on `serviceName` configuration: do we want to keep it in two places?
> 2. auth.to.local config name is a bit opaque, is there a better one?
> 3. Implement or remove SASL_KAFKA_SERVER_REALM config
> 4. Consider making Login's thread a daemon thread
> 5. Write test that shows authentication failure due to invalid user
> 6. Write test that shows authentication failure due to wrong password
> 7. Write test that shows authentication failure due ticket expiring



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

Re: [DISCUSS] KIP-36 - Rack aware replica assignment

[Allen Wang]

For 2 and 3, I have updated the KIP. Please take a look. One thing I have
changed is removing the proposal to add rack to TopicMetadataResponse. The
reason is that unlike UpdateMetadataRequest, TopicMetadataResponse does not
understand version. I don't see a way to include rack without breaking old
version of clients. That's probably why secure protocol is not included in
the TopicMetadataResponse either. I think it will be a much bigger change
to include rack in TopicMetadataResponse.

For 1, my concern is that doing rack aware assignment without complete
broker to rack mapping will result in assignment that is not rack aware and
fail to provide fault tolerance in the event of rack outage. This kind of
problem will be difficult to surface. And the cost of this problem is high:
you have to do partition reassignment if you are lucky to spot the problem
early on or face the consequence of data loss during real rack outage.

I do see the concern of fail-fast as it might also cause data loss if
producer is not able produce the message due to topic creation failure. Is
it feasible to treat dynamic topic creation and command tools differently?
We allow dynamic topic creation with incomplete broker-rack mapping and
fail fast in command line. Another option is to let user determine the
behavior for command line. For example, by default fail fast in command
line but allow incomplete broker-rack mapping if another switch is provided.




On Tue, Oct 20, 2015 at 10:05 AM, Aditya Auradkar <
aaurad...@linkedin.com.invalid> wrote:

> Hey Allen,
>
> 1. If we choose fail fast topic creation, we will have topic creation
> failures while upgrading the cluster. I really doubt we want this behavior.
> Ideally, this should be invisible to clients of a cluster. Currently, each
> broker is effectively its own rack. So we probably can use the rack
> information whenever possible but not make it a hard requirement. To extend
> Gwen's example, one badly configured broker should not degrade topic
> creation for the entire cluster.
>
> 2. Upgrade scenario - Can you add a section on the upgrade piece to confirm
> that old clients will not see errors? I believe ZookeeperConsumerConnector
> reads the Broker objects from ZK. I wanted to confirm that this will not
> cause any problems.
>
> 3. Could you elaborate your proposed changes to the UpdateMetadataRequest
> in the "Public Interfaces" section? Personally, I find this format easy to
> read in terms of wire protocol changes:
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-4+-+Command+line+and+centralized+administrative+operations#KIP-4-Commandlineandcentralizedadministrativeoperations-CreateTopicRequest
>
> Aditya
>
> On Fri, Oct 16, 2015 at 3:45 PM, Allen Wang  wrote:
>
> > KIP is updated include rack as an optional property for broker. Please
> take
> > a look and let me know if more details are needed.
> >
> > For the case where some brokers have rack and some do not, the current
> KIP
> > uses the fail-fast behavior. If there are concerns, we can further
> discuss
> > this in the email thread or next hangout.
> >
> >
> >
> > On Thu, Oct 15, 2015 at 10:42 AM, Allen Wang 
> wrote:
> >
> > > That's a good question. I can think of three actions if the rack
> > > information is incomplete:
> > >
> > > 1. Treat the node without rack as if it is on its unique rack
> > > 2. Disregard all rack information and fallback to current algorithm
> > > 3. Fail-fast
> > >
> > > Now I think about it, one and three make more sense. The reason for
> > > fail-fast is that user mistake for not providing the rack may never be
> > > found if we tolerate that and the assignment may not be rack aware as
> the
> > > user has expected and this creates debug problems when things fail.
> > >
> > > What do you think? If not fail-fast, is there anyway we can make the
> user
> > > error standing out?
> > >
> > >
> > > On Thu, Oct 15, 2015 at 10:17 AM, Gwen Shapira 
> > wrote:
> > >
> > >> Thanks! Just to clarify, when some brokers have rack assignment and
> some
> > >> don't, do we act like none of them have it? or like those without
> > >> assignment are in their own rack?
> > >>
> > >> The first scenario is good when first setting up rack-awareness, but
> the
> > >> second makes more sense for on-going maintenance (I can totally see
> > >> someone
> > >> adding a node and forgetting to set the rack property, we don't want
> > this
> > >> to change behavior for anything except the new node).
> > >>
> > >> What do you think?
> > >>
> > >> Gwen
> > >>
> > >> On Thu, Oct 15, 2015 at 10:13 AM, Allen Wang 
> > >> wrote:
> > >>
> > >> > For scenario 1:
> > >> >
> > >> > - Add the rack information to broker property file or dynamically
> set
> > >> it in
> > >> > the wrapper code to bootstrap Kafka server. You would do that for
> all
> > >> > brokers and restart the brokers one by one.
> > >> >
> > >> > In this scenario, the 

[GitHub] kafka pull request: Restore `SslConsumerTest` which was accidental...

[ijuma]

GitHub user ijuma opened a pull request:

https://github.com/apache/kafka/pull/350

Restore `SslConsumerTest` which was accidentally deleted in client-side 
assignment commit

Probably happened while resolving conflicts, commit: 
86eb74d9236c586af5889fe79f4b9e066c9c2af3

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/ijuma/kafka restore-ssl-consumer-test

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/350.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #350


commit 69c82cbe61ac19c5a86a1e76256315079644edf3
Author: Ismael Juma 
Date:   2015-10-22T09:50:30Z

Restore `SslConsumerTest` which was accidentally deleted in client-side 
assignment commit

Probably happened while resolving conflicts, commit: 
86eb74d9236c586af5889fe79f4b9e066c9c2af3




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-2017) Persist Coordinator State for Coordinator Failover

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-2017:
-

[~hachikuji] That makes sense. Persisting consumer metadata in Kafka makes it 
extendable in the future. I also agree with [~junrao] that it would work in 
most cases to simply skip generation id check and record the generation id if 
there is no information for the group.

> Persist Coordinator State for Coordinator Failover
> --
>
> Key: KAFKA-2017
> URL: https://issues.apache.org/jira/browse/KAFKA-2017
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: Onur Karaman
>Assignee: Guozhang Wang
> Fix For: 0.9.0.0
>
> Attachments: KAFKA-2017.patch, KAFKA-2017_2015-05-20_09:13:39.patch, 
> KAFKA-2017_2015-05-21_19:02:47.patch
>
>
> When a coordinator fails, the group membership protocol tries to failover to 
> a new coordinator without forcing all the consumers rejoin their groups. This 
> is possible if the coordinator persists its state so that the state can be 
> transferred during coordinator failover. This state consists of most of the 
> information in GroupRegistry and ConsumerRegistry.



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

Build failed in Jenkins: kafka-trunk-jdk8 #51

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-2686: Reset needsPartitionAssignment in 
SubscriptionState.assign()

--
[...truncated 5267 lines...]

kafka.coordinator.MemberMetadataTest > testVoteRaisesOnNoSupportedProtocols 
PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[0] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[1] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[2] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[3] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[4] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[5] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[6] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[7] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[8] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[9] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[10] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[11] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[12] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[13] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[14] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[15] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[16] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[17] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[18] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[19] PASSED

kafka.log.OffsetMapTest > testClear PASSED

kafka.log.OffsetMapTest > testBasicValidation PASSED

kafka.log.LogManagerTest > testCleanupSegmentsToMaintainSize PASSED

kafka.log.LogManagerTest > testRecoveryDirectoryMappingWithRelativeDirectory 
PASSED

kafka.log.LogManagerTest > testGetNonExistentLog PASSED

kafka.log.LogManagerTest > testTwoLogManagersUsingSameDirFails PASSED

kafka.log.LogManagerTest > testLeastLoadedAssignment PASSED

kafka.log.LogManagerTest > testCleanupExpiredSegments PASSED

kafka.log.LogManagerTest > testCheckpointRecoveryPoints PASSED

kafka.log.LogManagerTest > testTimeBasedFlush PASSED

kafka.log.LogManagerTest > testCreateLog PASSED

kafka.log.LogManagerTest > testRecoveryDirectoryMappingWithTrailingSlash PASSED

kafka.log.LogSegmentTest > testRecoveryWithCorruptMessage PASSED

kafka.log.LogSegmentTest > testRecoveryFixesCorruptIndex PASSED

kafka.log.LogSegmentTest > testReadFromGap PASSED

kafka.log.LogSegmentTest > testTruncate PASSED

kafka.log.LogSegmentTest > testReadBeforeFirstOffset PASSED

kafka.log.LogSegmentTest > testCreateWithInitFileSizeAppendMessage PASSED

kafka.log.LogSegmentTest > testChangeFileSuffixes PASSED

kafka.log.LogSegmentTest > testMaxOffset PASSED

kafka.log.LogSegmentTest > testNextOffsetCalculation PASSED

kafka.log.LogSegmentTest > testReadOnEmptySegment PASSED

kafka.log.LogSegmentTest > testReadAfterLast PASSED

kafka.log.LogSegmentTest > testCreateWithInitFileSizeClearShutdown PASSED

kafka.log.LogSegmentTest > testTruncateFull PASSED

kafka.log.FileMessageSetTest > testTruncate PASSED

kafka.log.FileMessageSetTest > testIterationOverPartialAndTruncation PASSED

kafka.log.FileMessageSetTest > testRead PASSED

kafka.log.FileMessageSetTest > testFileSize PASSED

kafka.log.FileMessageSetTest > testIteratorWithLimits PASSED

kafka.log.FileMessageSetTest > testPreallocateTrue PASSED

kafka.log.FileMessageSetTest > testIteratorIsConsistent PASSED

kafka.log.FileMessageSetTest > testIterationDoesntChangePosition PASSED

kafka.log.FileMessageSetTest > testWrittenEqualsRead PASSED

kafka.log.FileMessageSetTest > testWriteTo PASSED

kafka.log.FileMessageSetTest > testPreallocateFalse PASSED

kafka.log.FileMessageSetTest > testPreallocateClearShutdown PASSED

kafka.log.FileMessageSetTest > testSearch PASSED

kafka.log.FileMessageSetTest > testSizeInBytes PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[0] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[1] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[2] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[3] PASSED

kafka.log.LogTest > testParseTopicPartitionNameForMissingTopic PASSED

kafka.log.LogTest > testIndexRebuild PASSED

kafka.log.LogTest > testLogRolls PASSED

kafka.log.LogTest > testMessageSizeCheck PASSED

kafka.log.LogTest > testAsyncDelete PASSED

kafka.log.LogTest > testReadOutOfRange PASSED

kafka.log.LogTest > testReadAtLogGap PASSED

kafka.log.LogTest > testTimeBasedLogRoll PASSED

kafka.log.LogTest > testLoadEmptyLog PASSED

kafka.log.LogTest > testMessageSetSizeCheck PASSED

kafka.log.LogTest > testIndexResizingAtTruncation PASSED

kafka.log.LogTest > testCompactedTopicConstraints PASSED

kafka.log.LogTest > 

[GitHub] kafka pull request: MINOR: follow-up to KAFKA-2464 for renaming/cl...

[hachikuji]

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/354

MINOR: follow-up to KAFKA-2464 for renaming/cleanup



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka KAFKA-2464

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/354.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #354


commit 6ba0c30d2126457fa0efe5f26974ce5544ea771a
Author: Jason Gustafson 
Date:   2015-10-23T03:15:05Z

KAFKA-2464: follow-up for renaming/cleanup




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-2464) Client-side assignment and group generalization

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2464:
---

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/354

MINOR: follow-up to KAFKA-2464 for renaming/cleanup



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka KAFKA-2464

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/354.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #354


commit 6ba0c30d2126457fa0efe5f26974ce5544ea771a
Author: Jason Gustafson 
Date:   2015-10-23T03:15:05Z

KAFKA-2464: follow-up for renaming/cleanup




> Client-side assignment and group generalization
> ---
>
> Key: KAFKA-2464
> URL: https://issues.apache.org/jira/browse/KAFKA-2464
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>Priority: Blocker
> Fix For: 0.9.0.0
>
>
> Add support for client-side assignment and generalization of join group 
> protocol as documented here: 
> https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Client-side+Assignment+Proposal.



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

[jira] [Resolved] (KAFKA-2686) unsubscribe() call leaves KafkaConsumer in invalid state for manual topic-partition assignment

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-2686.
--
   Resolution: Fixed
Fix Version/s: 0.9.0.0

Issue resolved by pull request 352
[https://github.com/apache/kafka/pull/352]

> unsubscribe() call leaves KafkaConsumer in invalid state for manual 
> topic-partition assignment
> --
>
> Key: KAFKA-2686
> URL: https://issues.apache.org/jira/browse/KAFKA-2686
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: The Data Lorax
>Assignee: Guozhang Wang
> Fix For: 0.9.0.0
>
>
> The bellow code snippet demonstrated the problem.
> Basically, the unsubscribe() call leaves the KafkaConsumer in a state that 
> means poll() will always return empty record sets, even if new 
> topic-partitions have been assigned that have messages pending.  This is 
> because unsubscribe() sets SubscriptionState.needsPartitionAssignment to 
> true, and assign() does not clear this flag. The only thing that clears this 
> flag is when the consumer handles the response from a JoinGroup request.
> {code}
> final KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
> ConsumerRecords records = consumer.poll(100);// <- Works, 
> returning records
> consumer.unsubscribe();   // Puts consumer into invalid state.
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 2)));
> records = consumer.poll(100);// <- Always returns empty record set.
> {code}



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

[GitHub] kafka pull request: KAFKA-2686: Reset needsPartitionAssignment in ...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/352


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-2686) unsubscribe() call leaves KafkaConsumer in invalid state for manual topic-partition assignment

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2686:
---

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/352


> unsubscribe() call leaves KafkaConsumer in invalid state for manual 
> topic-partition assignment
> --
>
> Key: KAFKA-2686
> URL: https://issues.apache.org/jira/browse/KAFKA-2686
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: The Data Lorax
>Assignee: Guozhang Wang
> Fix For: 0.9.0.0
>
>
> The bellow code snippet demonstrated the problem.
> Basically, the unsubscribe() call leaves the KafkaConsumer in a state that 
> means poll() will always return empty record sets, even if new 
> topic-partitions have been assigned that have messages pending.  This is 
> because unsubscribe() sets SubscriptionState.needsPartitionAssignment to 
> true, and assign() does not clear this flag. The only thing that clears this 
> flag is when the consumer handles the response from a JoinGroup request.
> {code}
> final KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
> ConsumerRecords records = consumer.poll(100);// <- Works, 
> returning records
> consumer.unsubscribe();   // Puts consumer into invalid state.
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 2)));
> records = consumer.poll(100);// <- Always returns empty record set.
> {code}



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

Re: Java high level consumer providing duplicate messages when auto commit is off

[Jiangjie Qin]

Hi Cliff,

If auto.offset.commit is set to true, the offset will be committed in
following cases in addition to periodical offset commit:

1. During consumer rebalance before release the partition ownership.
If consumer A owns partition P before rebalance, it will commit offset for
partition P during rebalance. If consumer B become the new owner of
partition P after rebalance, it will start from the committed offset, so
there will be no duplicate messages.
2. When consumer closes.

Rebalance will be triggered in the following cases:
1. A consumer joins/leaves the group.
2. Some topic/partition changes occurred to the interested topics.(e.g.
partition expansion for a topic; a new topic created and the consumer is
using a wildcard that matches the new topic name)

To answer your question:
Simple consumer should not interfere with high level consumer because it
does not have any group management embedded.

Typically a single high level consumer group will not rebalance unless
there is topic/partition change. However, it is possible the consumer
itself dropped out of the group and rejoins. This typically happens when
you have a ZK session timeout. In that case, you should see "ZK expired" in
your log. You can search for that and see if that is the problem.

Jiangjie (Becket) Qin


On Thu, Oct 22, 2015 at 1:14 PM, Cliff Rhyne  wrote:

> We did some more testing with logging turned on (I figured out why it
> wasn't working).  We tried increasing the JVM memory capacity on our test
> server (it's lower than in production) and increasing the zookeeper
> timeouts.  Neither changed the results.  With trace logging enabled, we saw
> that we were getting rebalances even though there is only one high level
> consumer running (there previously was a simple consumer that was told to
> disconnect, but that consumer only checked the offsets and never consumed
> data).
>
> - Is there possibly a race condition where the simple consumer has a hold
> on a partition and shutdown is called before starting a high level consumer
> but shutdown is done asynchronously?
> - What are the various things that can cause a consumer rebalance other
> than adding / removing high level consumers?
>
> Thanks,
> Cliff
>
> On Wed, Oct 21, 2015 at 4:20 PM, Cliff Rhyne  wrote:
>
> > Hi Kris,
> >
> > Thanks for the tip.  I'm going to investigate this further.  I checked
> and
> > we have fairly short zk timeouts and run with a smaller memory allocation
> > on the two environments we encounter this issue.  I'll let you all know
> > what I find.
> >
> > I saw this ticket https://issues.apache.org/jira/browse/KAFKA-2049 that
> > seems to be related to the problem (but would only inform that an issue
> > occurred).  Are there any other open issues that could be worked on to
> > improve Kafka's handling of this situation?
> >
> > Thanks,
> > Cliff
> >
> > On Wed, Oct 21, 2015 at 2:53 PM, Kris K  wrote:
> >
> >> Hi Cliff,
> >>
> >> One other case I observed in my environment is - when there were gc
> pauses
> >> on one of our high level consumer in the group.
> >>
> >> Thanks,
> >> Kris
> >>
> >> On Wed, Oct 21, 2015 at 10:12 AM, Cliff Rhyne  wrote:
> >>
> >> > Hi James,
> >> >
> >> > There are two scenarios we run:
> >> >
> >> > 1. Multiple partitions with one consumer per partition.  This rarely
> has
> >> > starting/stopping of consumers, so the pool is very static.  There is
> a
> >> > configured consumer timeout, which is causing the
> >> ConsumerTimeoutException
> >> > to get thrown prior to the test starting.  We handle this exception
> and
> >> > then resume consuming.
> >> > 2. Single partition with one consumer.  This consumer is started by a
> >> > triggered condition (number of messages pending to be processed in the
> >> > kafka topic or a schedule).  The consumer is stopped after processing
> is
> >> > completed.
> >> >
> >> > In both cases, based on my understanding there shouldn't be a
> rebalance
> >> as
> >> > either a) all consumers are running or b) there's only one consumer /
> >> > partition.  Also, the same consumer group is used by all consumers in
> >> > scenario 1 and 2.  Is there a good way to investigate whether
> rebalances
> >> > are occurring?
> >> >
> >> > Thanks,
> >> > Cliff
> >> >
> >> > On Wed, Oct 21, 2015 at 11:37 AM, James Cheng 
> wrote:
> >> >
> >> > > Do you have multiple consumers in a consumer group?
> >> > >
> >> > > I think that when a new consumer joins the consumer group, that the
> >> > > existing consumers will stop consuming during the group rebalance,
> and
> >> > then
> >> > > when they start consuming again, that they will consume from the
> last
> >> > > committed offset.
> >> > >
> >> > > You should get more verification on this, tho. I might be
> remembering
> >> > > wrong.
> >> > >
> >> > > -James
> >> > >
> >> > > > On Oct 21, 2015, at 8:40 AM, Cliff Rhyne 
> wrote:
> >> > > >
> >> > 

[GitHub] kafka pull request: KAFKA-2338: add force option to topic / config...

[benstopford]

GitHub user benstopford opened a pull request:

https://github.com/apache/kafka/pull/351

KAFKA-2338: add force option to topic / config command so they can be 
called programatically

Tiny change to add a force option to the topic and config commands so they 
can be called programatically without requiring user input. 

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/benstopford/kafka CPKAFKA-61B

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/351.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #351


commit b29316e7174f21a91043d7a3aa451b6345324dd2
Author: Ben Stopford 
Date:   2015-10-22T12:27:13Z

KAFKA-2338: add 'force' option to avoid console prompts




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-2675) SASL/Kerberos follow-up

[Ismael Juma (JIRA)]

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

Ismael Juma commented on KAFKA-2675:


Some input would be appreciated on the following:

1. Decide on `serviceName` configuration: do we want to keep it in two places?

This is the only config we are storing in 2 places (all the other ones are 
either in the Kafka config or the JAAS file). We need to have it in the Kafka 
config because the IBM JDK fails if it sees this property in the JAAS file. 
Given this, I think we need strong justification to support setting this in the 
JAAS file. [~harsha_ch], you said that the reason to have it in the JAAS file 
is that many other projects do that. Do you have a reference for the projects 
that do it?

2. auth.to.local config name is a bit opaque, is there a better one?

The full config name is kafka.security.auth.to.local, and it's similar to the 
Hadoop system property:

http://www.cloudera.com/content/www/en-us/documentation/archive/cdh/4-x/4-3-1/CDH4-Security-Guide/cdh4sg_topic_19.html

Other options that I can think of could be:
principal.to.local
principal.to.local.rules
auth.to.local.rules

A reason to have `rules` in the name is that we could support the definition of 
a class to do the mapping in the future. Hortonworks supports this via a 
`principal.to.local.class` property:

http://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.3.0/bk_secure-kafka-ambari/content/ch_secure-kafka-config-options.html

3. Implement or remove SASL_KAFKA_SERVER_REALM config

This config is not used at the moment. The documentation says "Default will be 
from kafka jaas config". Maybe we can remove it then? [~harsha_ch], your input 
is appreciated.

4. Consider making Login's thread a daemon thread

Rajini asked if this should be a daemon thread. It was a daemon thread to start 
with, but [~junrao] suggested that it perhaps should be non-daemon because we 
wait for it during shutdown. [~rsivaram]'s concern is that we may prevent 
clients from exiting normally. This won't happen if `KafkaConsumer.close()` is 
called, but it can happen if `close()` is not called. It seems like making it a 
daemon thread is the safest option. What do you think [~junrao]?

It would be very helpful to get your feedback so that we can wrap the 
SASL/Kerberos feature.

> SASL/Kerberos follow-up
> ---
>
> Key: KAFKA-2675
> URL: https://issues.apache.org/jira/browse/KAFKA-2675
> Project: Kafka
>  Issue Type: Sub-task
>  Components: security
>Reporter: Ismael Juma
>Assignee: Ismael Juma
> Fix For: 0.9.0.0
>
>
> This is a follow-up to KAFKA-1686. 
> 1. Decide on `serviceName` configuration: do we want to keep it in two places?
> 2. auth.to.local config name is a bit opaque, is there a better one?
> 3. Implement or remove SASL_KAFKA_SERVER_REALM config
> 4. Consider making Login's thread a daemon thread
> 5. Write test that shows authentication failure due to invalid user
> 6. Write test that shows authentication failure due to wrong password
> 7. Write test that shows authentication failure due ticket expiring



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

[jira] [Commented] (KAFKA-2338) Warn users if they change max.message.bytes that they also need to update broker and consumer settings

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2338:
---

GitHub user benstopford opened a pull request:

https://github.com/apache/kafka/pull/351

KAFKA-2338: add force option to topic / config command so they can be 
called programatically

Tiny change to add a force option to the topic and config commands so they 
can be called programatically without requiring user input. 

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/benstopford/kafka CPKAFKA-61B

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/351.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #351


commit b29316e7174f21a91043d7a3aa451b6345324dd2
Author: Ben Stopford 
Date:   2015-10-22T12:27:13Z

KAFKA-2338: add 'force' option to avoid console prompts




> Warn users if they change max.message.bytes that they also need to update 
> broker and consumer settings
> --
>
> Key: KAFKA-2338
> URL: https://issues.apache.org/jira/browse/KAFKA-2338
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.8.2.1
>Reporter: Ewen Cheslack-Postava
>Assignee: Ben Stopford
>Priority: Critical
> Fix For: 0.9.0.0
>
> Attachments: KAFKA-2338.patch, KAFKA-2338_2015-07-18_00:37:31.patch, 
> KAFKA-2338_2015-07-21_13:21:19.patch, KAFKA-2338_2015-08-24_14:32:38.patch, 
> KAFKA-2338_2015-09-02_19:27:17.patch
>
>
> We already have KAFKA-1756 filed to more completely address this issue, but 
> it is waiting for some other major changes to configs to completely protect 
> users from this problem.
> This JIRA should address the low hanging fruit to at least warn users of the 
> potential problems. Currently the only warning is in our documentation.
> 1. Generate a warning in the kafka-topics.sh tool when they change this 
> setting on a topic to be larger than the default. This needs to be very 
> obvious in the output.
> 2. Currently, the broker's replica fetcher isn't logging any useful error 
> messages when replication can't succeed because a message size is too large. 
> Logging an error here would allow users that get into a bad state to find out 
> why it is happening more easily. (Consumers should already be logging a 
> useful error message.)



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

[GitHub] kafka pull request: KAFKA-2671: Enable starting Kafka server with ...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/330


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-2671) Enable starting Kafka server with a Properties object

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2671:
---

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/330


> Enable starting Kafka server with a Properties object
> -
>
> Key: KAFKA-2671
> URL: https://issues.apache.org/jira/browse/KAFKA-2671
> Project: Kafka
>  Issue Type: Improvement
>  Components: core
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.9.0.0
>
>
> Kafka, as of now, can only be started with a properties file and override 
> params. It makes life easier for management applications to be able to start 
> Kafka with a properties object programatically.
> The changes required to enable this are minimal, just a tad bit of 
> refactoring of kafka.Kafka. The changes must maintain current behavior intact.



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

[jira] [Resolved] (KAFKA-2671) Enable starting Kafka server with a Properties object

[Gwen Shapira (JIRA)]

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

Gwen Shapira resolved KAFKA-2671.
-
   Resolution: Fixed
Fix Version/s: 0.9.0.0

Issue resolved by pull request 330
[https://github.com/apache/kafka/pull/330]

> Enable starting Kafka server with a Properties object
> -
>
> Key: KAFKA-2671
> URL: https://issues.apache.org/jira/browse/KAFKA-2671
> Project: Kafka
>  Issue Type: Improvement
>  Components: core
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.9.0.0
>
>
> Kafka, as of now, can only be started with a properties file and override 
> params. It makes life easier for management applications to be able to start 
> Kafka with a properties object programatically.
> The changes required to enable this are minimal, just a tad bit of 
> refactoring of kafka.Kafka. The changes must maintain current behavior intact.



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

[jira] [Work started] (KAFKA-2684) Add force option to TopicCommand & ConfigCommand to suppress console prompts

[Ben Stopford (JIRA)]

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

Work on KAFKA-2684 started by Ben Stopford.
---
> Add force option to TopicCommand & ConfigCommand to suppress console prompts
> 
>
> Key: KAFKA-2684
> URL: https://issues.apache.org/jira/browse/KAFKA-2684
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ben Stopford
>Assignee: Ben Stopford
>Priority: Trivial
>
> Add force option to TopicCommand & Config Command to suppress console prompts
> This is useful from system tests etc which call these scripts 
> programmatically. 
> Relates to KAFKA-2338



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

[jira] [Updated] (KAFKA-2684) Add force option to TopicCommand & ConfigCommand to suppress console prompts

[Ben Stopford (JIRA)]

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

Ben Stopford updated KAFKA-2684:

Description: 
Add force option to TopicCommand & Config Command to suppress console prompts

This is useful from system tests etc which call these scripts programmatically. 

Relates to KAFKA-2338


  was:
Add force option to TopicCommand & Config Command to suppress console prompts

This is useful from system tests etc which call these scripts programmatically. 




> Add force option to TopicCommand & ConfigCommand to suppress console prompts
> 
>
> Key: KAFKA-2684
> URL: https://issues.apache.org/jira/browse/KAFKA-2684
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ben Stopford
>Priority: Trivial
>
> Add force option to TopicCommand & Config Command to suppress console prompts
> This is useful from system tests etc which call these scripts 
> programmatically. 
> Relates to KAFKA-2338



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

Re: [DISCUSS] KIP-38: ZooKeeper Authentication

[Flavio Junqueira]

Ok, thanks for spotting it.

-Flavio

> On 22 Oct 2015, at 05:54, Jun Rao  wrote:
> 
> It seems that in the secure -> unsecure plan, step 3 needs to be done
> before step 2.
> 
> Thanks,
> 
> Jun
> 
> On Wed, Oct 21, 2015 at 3:59 PM, Flavio Junqueira  wrote:
> 
>> Ok, thanks for the feedback, Todd. I have updated the KIP with some of the
>> points discussed here. There is more to add based on these last comments,
>> though.
>> 
>> -Flavio
>> 
>>> On 21 Oct 2015, at 23:43, Todd Palino  wrote:
>>> 
>>> On Wed, Oct 21, 2015 at 3:38 PM, Flavio Junqueira > > wrote:
>>> 
 
> On 21 Oct 2015, at 21:54, Todd Palino  wrote:
> 
> Thanks for the clarification on that, Jun. Obviously, we haven't been
 doing
> much with ZK authentication around here yet. There is still a small
 concern
> there, mostly in that you should not share credentials any more than is
> necessary, which would argue for being able to use a different ACL than
 the
> default. I don't really like the idea of having to use the exact same
> credentials for executing the admin tools as we do for running the
 brokers.
> Given that we don't need to share the credentials with all consumers, I
> think we can work around it.
> 
 
 Let me add that a feature to separate the sub-trees of users sharing an
 ensemble is chroot.
 
 On different credentials for admin tools, this sounds doable by setting
 the ACLs of znodes. For example, there could be an admin id and a broker
 id, both with the ability of changing znodes, but different credentials.
 Would something like that work for you?
 
>>> 
>>> It would be a nice option to have, as the credentials can be protected
>>> differently. I would consider this a nice to have, and not an "absolutely
>>> must have" feature at this point.
>>> 
>>> 
 This does bring up another good question, however. What will be the
 process
> for having to rotate the credentials? That is, if the credentials are
> compromised and need to be changed, how can that be accomplished with
>> the
> cluster online. I'm guessing some combination of using skipAcl on the
> Zookeeper ensemble and config changes to the brokers will be required,
 but
> this is an important enough operation that we should make sure it's
> reasonable to perform and that it is documented.
 
 Right now there is no kafka support in the plan for this. But this is
 doable directly through the zk api. Would it be sufficient to write down
 how to perform such an operation via the zk api or do we need a tool to
>> do
 it?
 
>>> 
>>> I think as long as there is a documented procedure for how to do it, that
>>> will be good enough. It's mostly about making sure that we can, and that
>> we
>>> don't put something in place that would require downtime to a cluster in
>>> order to change credentials. We can always develop a tool later if it is
>> a
>>> requested item.
>>> 
>>> Thanks!
>>> 
>>> -Todd
>>> 
>>> 
>>> 
 
 -Flavio
 
> 
> 
> On Wed, Oct 21, 2015 at 1:23 PM, Jun Rao  wrote:
> 
>> Parth,
>> 
>> For 2), in your approach, the broker/controller will then always have
 the
>> overhead of resetting the ACL on startup after zookeeper.set.acl is
>> set
 to
>> true. The benefit of using a separate migration tool is that you paid
 the
>> cost only once during upgrade. It is an extra step during the upgrade.
>> However, given the other things that you need to do to upgrade to
>> 0.9.0
>> (e.g. two rounds of rolling upgrades on all brokers, etc), I am not
 sure if
>> it's worth to optimize away of this step. We probably just need to
 document
>> this clearly.
>> 
>> Todd,
>> 
>> Just to be clear about the shared ZK usage. Once you set
 CREATOR_ALL_ACL +
>> READ_ACL_UNSAFE on a path, only ZK clients with the same user as the
>> creator can modify the path. Other ZK clients authenticated with a
>> different user can read, but not modify the path. Are you concerned
 about
>> the reads or the writes to ZK?
>> 
>> Thanks,
>> 
>> Jun
>> 
>> 
>> 
>> On Wed, Oct 21, 2015 at 10:46 AM, Flavio Junqueira 
 wrote:
>> 
>>> 
 On 21 Oct 2015, at 18:07, Parth Brahmbhatt <
>> pbrahmbh...@hortonworks.com>
>>> wrote:
 
 I have 2 suggestions:
 
 1) We need to document how does one move from secure to non secure
 environment:
1) change the config on all brokers to zookeeper.set.acl = false
>>> and do a
 rolling upgrade.
2) Run the migration script with the jass config file so it is
>> sasl
 authenticated with zookeeper and change 

[jira] [Assigned] (KAFKA-2684) Add force option to TopicCommand & ConfigCommand to suppress console prompts

[Ben Stopford (JIRA)]

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

Ben Stopford reassigned KAFKA-2684:
---

Assignee: Ben Stopford

> Add force option to TopicCommand & ConfigCommand to suppress console prompts
> 
>
> Key: KAFKA-2684
> URL: https://issues.apache.org/jira/browse/KAFKA-2684
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ben Stopford
>Assignee: Ben Stopford
>Priority: Trivial
>
> Add force option to TopicCommand & Config Command to suppress console prompts
> This is useful from system tests etc which call these scripts 
> programmatically. 
> Relates to KAFKA-2338



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

[jira] [Created] (KAFKA-2684) Add force option to TopicCommand & ConfigCommand to suppress console prompts

[Ben Stopford (JIRA)]

Ben Stopford created KAFKA-2684:
---

 Summary: Add force option to TopicCommand & ConfigCommand to 
suppress console prompts
 Key: KAFKA-2684
 URL: https://issues.apache.org/jira/browse/KAFKA-2684
 Project: Kafka
  Issue Type: Bug
Reporter: Ben Stopford
Priority: Trivial


Add force option to TopicCommand & Config Command to suppress console prompts

This is useful from system tests etc which call these scripts programmatically. 





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

[jira] [Updated] (KAFKA-2684) Add force option to TopicCommand & ConfigCommand to suppress console prompts

[Ben Stopford (JIRA)]

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

Ben Stopford updated KAFKA-2684:

Status: Patch Available  (was: In Progress)

> Add force option to TopicCommand & ConfigCommand to suppress console prompts
> 
>
> Key: KAFKA-2684
> URL: https://issues.apache.org/jira/browse/KAFKA-2684
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ben Stopford
>Assignee: Ben Stopford
>Priority: Trivial
>
> Add force option to TopicCommand & Config Command to suppress console prompts
> This is useful from system tests etc which call these scripts 
> programmatically. 
> Relates to KAFKA-2338



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

[jira] [Created] (KAFKA-2685) "alter topic" on non-existent topic exits without error

[Gwen Shapira (JIRA)]

Gwen Shapira created KAFKA-2685:
---

 Summary: "alter topic" on non-existent topic exits without error
 Key: KAFKA-2685
 URL: https://issues.apache.org/jira/browse/KAFKA-2685
 Project: Kafka
  Issue Type: Bug
Reporter: Gwen Shapira


When running:
kafka-topics --zookeeper localhost:2181 --alter --topic test --config 
unclean.leader.election.enable=false

and topic "test" does not exist, the command simply return with no error 
message.

We expect to see an error when trying to modify non-existing topics, so user 
will have a chance to catch and correct typos.



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

[jira] [Updated] (KAFKA-2686) unsubscribe() call leaves KafkaConsumer in invalid state for manual topic-partition assignment

[The Data Lorax (JIRA)]

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

The Data Lorax updated KAFKA-2686:
--
Description: 
The bellow code snippet demonstrated the problem.

Basically, the unsubscribe() call leaves the KafkaConsumer in a state that 
means poll() will always return empty record sets, even if new topic-partitions 
have been assigned that have messages pending.  This is because unsubscribe() 
sets SubscriptionState.needsPartitionAssignment to true, and assign() does not 
clear this flag. The only thing that clears this flag is when the consumer 
handles the response from a JoinGroup request.

{code}
final KafkaConsumer consumer = new KafkaConsumer<>(props);

consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
ConsumerRecords records = consumer.poll(100);// <- Works, 
returning records

consumer.unsubscribe();   // Puts consumer into invalid state.

consumer.assign(Collections.singletonList(new TopicPartition(topicName, 2)));
records = consumer.poll(100);// <- Always returns empty record set.
{code}



> unsubscribe() call leaves KafkaConsumer in invalid state for manual 
> topic-partition assignment
> --
>
> Key: KAFKA-2686
> URL: https://issues.apache.org/jira/browse/KAFKA-2686
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: The Data Lorax
>Assignee: Neha Narkhede
>
> The bellow code snippet demonstrated the problem.
> Basically, the unsubscribe() call leaves the KafkaConsumer in a state that 
> means poll() will always return empty record sets, even if new 
> topic-partitions have been assigned that have messages pending.  This is 
> because unsubscribe() sets SubscriptionState.needsPartitionAssignment to 
> true, and assign() does not clear this flag. The only thing that clears this 
> flag is when the consumer handles the response from a JoinGroup request.
> {code}
> final KafkaConsumer consumer = new KafkaConsumer<>(props);
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
> ConsumerRecords records = consumer.poll(100);// <- Works, 
> returning records
> consumer.unsubscribe();   // Puts consumer into invalid state.
> consumer.assign(Collections.singletonList(new TopicPartition(topicName, 2)));
> records = consumer.poll(100);// <- Always returns empty record set.
> {code}



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

[GitHub] kafka pull request: MINOR: Restore `SslConsumerTest` which was acc...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/350


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] kafka pull request: KAFKA-2678; partition level lag metrics can be...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/346


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Created] (KAFKA-2686) unsubscribe() call leaves KafkaConsumer in invalid state for manual topic-partition assignment

[The Data Lorax (JIRA)]

The Data Lorax created KAFKA-2686:
-

 Summary: unsubscribe() call leaves KafkaConsumer in invalid state 
for manual topic-partition assignment
 Key: KAFKA-2686
 URL: https://issues.apache.org/jira/browse/KAFKA-2686
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Affects Versions: 0.9.0.0
Reporter: The Data Lorax
Assignee: Neha Narkhede






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

[jira] [Resolved] (KAFKA-2678) partition level lag metrics can be negative

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-2678.
--
   Resolution: Fixed
Fix Version/s: 0.9.0.0

Issue resolved by pull request 346
[https://github.com/apache/kafka/pull/346]

> partition level lag metrics can be negative
> ---
>
> Key: KAFKA-2678
> URL: https://issues.apache.org/jira/browse/KAFKA-2678
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.8.2.1
>Reporter: Jun Rao
>Assignee: Dong Lin
> Fix For: 0.9.0.0
>
>
> Currently, the per partition level lag metric can be negative since the last 
> committed offset can be smaller than the follower's offset. This is a bit 
> confusing to end users. We probably should lower bound it by 0.



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

[jira] [Commented] (KAFKA-2678) partition level lag metrics can be negative

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2678:
---

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/346


> partition level lag metrics can be negative
> ---
>
> Key: KAFKA-2678
> URL: https://issues.apache.org/jira/browse/KAFKA-2678
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.8.2.1
>Reporter: Jun Rao
>Assignee: Dong Lin
> Fix For: 0.9.0.0
>
>
> Currently, the per partition level lag metric can be negative since the last 
> committed offset can be smaller than the follower's offset. This is a bit 
> confusing to end users. We probably should lower bound it by 0.



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