Jenkins build is back to normal : kafka-trunk-jdk8 #428

[Apache Jenkins Server]

See 

Re: Protocol Documentation WIP

[Ewen Cheslack-Postava]

Agreed w/ Jay's comment on the PR that this is a great idea. What if we
included this as is, but instead of putting it all on the docs page we put
it on its own page? This:

1. Gets us something useful wrt protocol docs immediately. As you point
out, this is already useful to a bunch of people.
2. Doesn't block on us figuring out a better way of managing docs (although
this may need adjustment once we choose a different format for docs).
3. Gets the code checked in so any requests for refinements around protocol
docs can often be resolved with a patch to the code for autogenerating them.

-Ewen

On Wed, Feb 24, 2016 at 11:57 PM, Grant Henke  wrote:

> We have gotten a lot of requests for updating the protocol documentation:
>
>- KAFKA-3124: Update protocol wiki page to reflect latest
>request/response formats
>- KAFKA-2865: Improve Request API Error Code Documentation
>- KAFKA-3099: Please update A+Guide+To+The+Kafka+Protocol Wiki
>- KAFKA-2495: Protocol guide only partially updated after
>ConsumerMetadata* addition?
>- KAFKA-2494: Document ReplicaId in OffsetRequest in the protocol guide
>
> Generating some docs to go either in the wiki or on the actual
> documentation site would definitely be valuable and ensure our protocol
> docs are up to date. Although the output could be fairly large, so I am not
> sure it should go on the main page of our "one page" documentation site, as
> the page is already fairly long to scroll.
>
> I made a very rough, super hacky, implementation that outputs some useful
> data in a similar format to the existing wiki. There is a pull request and
> a gist of the output for anyone that it could help in the mean time (links
> below).  In the patch I listed some "todo" comments for things I would
> still like to see in the output. I don't have time to polish up the patch
> to completion now, due to other priorities.
>
> Is anyone interested in finishing the patch? If so please feel free to use
> or ignore the code shared.
>
>- Pull request: https://github.com/apache/kafka/pull/970
>- Gist of output:
>
> https://gist.github.com/granthenke/dae9e60b5e548acead58#file-kafka_protocol-html
>- In browser preview of Gist:
>
> https://rawgit.com/granthenke/dae9e60b5e548acead58/raw/a2da4279d6fd11e8056927695c7c091e58607f38/kafka_protocol.html
>
> Thanks,
> Grant
> --
> Grant Henke
> Software Engineer | Cloudera
> gr...@cloudera.com | twitter.com/gchenke | linkedin.com/in/granthenke
>



-- 
Thanks,
Ewen

[jira] [Commented] (KAFKA-3279) SASL implementation checks for unused System property java.security.auth.login.config

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> SASL implementation checks for unused System property 
> java.security.auth.login.config
> -
>
> Key: KAFKA-3279
> URL: https://issues.apache.org/jira/browse/KAFKA-3279
> Project: Kafka
>  Issue Type: Bug
>  Components: security
>Reporter: Rajini Sivaram
>Assignee: Rajini Sivaram
> Fix For: 0.10.0.0
>
>
> In many environments (eg. JEE containers), JAAS configuration may be set 
> using methods different from the System property 
> {{java.security.auth.login.config}}. While Kafka obtains JAAS configuration 
> correctly using {{Configuration.getConfiguration()}},  an exception is thrown 
> if the System property {{java.security.auth.login.config}} is not set even 
> when the property is never used. There are also misleading error messages 
> which refer to the value of this property which may or may not be the 
> configuration for which the error is being reported. 



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

[GitHub] kafka pull request: KAFKA-3279: Remove checks for JAAS system prop...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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] [Updated] (KAFKA-3279) SASL implementation checks for unused System property java.security.auth.login.config

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-3279:
-
Resolution: Fixed
Status: Resolved  (was: Patch Available)

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

> SASL implementation checks for unused System property 
> java.security.auth.login.config
> -
>
> Key: KAFKA-3279
> URL: https://issues.apache.org/jira/browse/KAFKA-3279
> Project: Kafka
>  Issue Type: Bug
>  Components: security
>Reporter: Rajini Sivaram
>Assignee: Rajini Sivaram
> Fix For: 0.10.0.0
>
>
> In many environments (eg. JEE containers), JAAS configuration may be set 
> using methods different from the System property 
> {{java.security.auth.login.config}}. While Kafka obtains JAAS configuration 
> correctly using {{Configuration.getConfiguration()}},  an exception is thrown 
> if the System property {{java.security.auth.login.config}} is not set even 
> when the property is never used. There are also misleading error messages 
> which refer to the value of this property which may or may not be the 
> configuration for which the error is being reported. 



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

[GitHub] kafka pull request: [MINOR] Fixed documenation of parameter "block...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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-3356) Remove ConsumerOffsetChecker, deprecated in 0.9, in 0.11

[Ashish K Singh (JIRA)]

Ashish K Singh created KAFKA-3356:
-

 Summary: Remove ConsumerOffsetChecker, deprecated in 0.9, in 0.11
 Key: KAFKA-3356
 URL: https://issues.apache.org/jira/browse/KAFKA-3356
 Project: Kafka
  Issue Type: Bug
  Components: tools
Affects Versions: 0.11.0.0
Reporter: Ashish K Singh
Assignee: Ashish K Singh
 Fix For: 0.11.0.0


ConsumerOffsetChecker is marked deprecated as of 0.9, should be removed in 0.11.



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

[jira] [Commented] (KAFKA-3355) GetOffsetShell command doesn't work with SASL enabled Kafka

[Ashish K Singh (JIRA)]

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

Ashish K Singh commented on KAFKA-3355:
---

[~xiaotao183] you are correct and this problem persists in following tools.

* ReplicationVerificationTool
* SimpleConsumerShell
* GetOffsetShell

As we are deprecating old producer, we should really move these tools over to 
use new producer. [~gwenshap] do you think this could be something worth 
considering for 0.10?

> GetOffsetShell command doesn't work with SASL enabled Kafka
> ---
>
> Key: KAFKA-3355
> URL: https://issues.apache.org/jira/browse/KAFKA-3355
> Project: Kafka
>  Issue Type: Bug
>  Components: tools
>Affects Versions: 0.9.0.1
> Environment: Kafka 0.9.0.1
>Reporter: TAO XIAO
>
> I found that GetOffsetShell doesn't work with SASL enabled Kafka. I believe 
> this is due to old producer being used in GetOffsetShell.
> Kafka version 0.9.0.1
> Exception
> % bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 
> localhost:9092 --topic test --time -1
> [2016-03-04 21:43:56,597] INFO Verifying properties 
> (kafka.utils.VerifiableProperties)
> [2016-03-04 21:43:56,613] INFO Property client.id is overridden to 
> GetOffsetShell (kafka.utils.VerifiableProperties)
> [2016-03-04 21:43:56,613] INFO Property metadata.broker.list is overridden to 
> localhost:9092 (kafka.utils.VerifiableProperties)
> [2016-03-04 21:43:56,613] INFO Property request.timeout.ms is overridden to 
> 1000 (kafka.utils.VerifiableProperties)
> [2016-03-04 21:43:56,674] INFO Fetching metadata from broker 
> BrokerEndPoint(0,localhost,9092) with correlation id 0 for 1 topic(s) 
> Set(test) (kafka.client.ClientUtils$)
> [2016-03-04 21:43:56,689] INFO Connected to localhost:9092 for producing 
> (kafka.producer.SyncProducer)
> [2016-03-04 21:43:56,705] WARN Fetching topic metadata with correlation id 0 
> for topics [Set(test)] from broker [BrokerEndPoint(0,localhost,9092)] failed 
> (kafka.client.ClientUtils$)
> java.nio.BufferUnderflowException
>   at java.nio.Buffer.nextGetIndex(Buffer.java:498)
>   at java.nio.HeapByteBuffer.getShort(HeapByteBuffer.java:304)
>   at kafka.api.ApiUtils$.readShortString(ApiUtils.scala:36)
>   at kafka.cluster.BrokerEndPoint$.readFrom(BrokerEndPoint.scala:52)
>   at 
> kafka.api.TopicMetadataResponse$$anonfun$1.apply(TopicMetadataResponse.scala:28)
>   at 
> kafka.api.TopicMetadataResponse$$anonfun$1.apply(TopicMetadataResponse.scala:28)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
>   at scala.collection.immutable.Range.foreach(Range.scala:166)
>   at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
>   at scala.collection.AbstractTraversable.map(Traversable.scala:104)
>   at 
> kafka.api.TopicMetadataResponse$.readFrom(TopicMetadataResponse.scala:28)
>   at kafka.producer.SyncProducer.send(SyncProducer.scala:120)
>   at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
>   at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
>   at kafka.tools.GetOffsetShell$.main(GetOffsetShell.scala:78)
>   at kafka.tools.GetOffsetShell.main(GetOffsetShell.scala)



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

[jira] [Updated] (KAFKA-3252) compression type for a topic should be used during log compaction

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3252:
--
Status: Patch Available  (was: Open)

> compression type for a topic should be used during log compaction 
> --
>
> Key: KAFKA-3252
> URL: https://issues.apache.org/jira/browse/KAFKA-3252
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.0
>Reporter: Jun Rao
>Assignee: Ashish K Singh
> Fix For: 0.10.0.0
>
>
> Currently, the broker uses the specified compression type in a topic for 
> newly published messages. However, during log compaction, it still uses the 
> compression codec in the original message. To be consistent, it seems that we 
> should use the compression type in a topic when copying the messages to new 
> log segments.



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

[jira] [Updated] (KAFKA-3247) kafka-topic tool can use an option to see just "unbalanced partitions"

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3247:
--
Status: Patch Available  (was: Open)

> kafka-topic tool can use an option to see just "unbalanced partitions"
> --
>
> Key: KAFKA-3247
> URL: https://issues.apache.org/jira/browse/KAFKA-3247
> Project: Kafka
>  Issue Type: Bug
>Reporter: Gwen Shapira
>Assignee: Ashish K Singh
>
> We already have an option for under replicated partitions, it will be nice to 
> also have the ability to just see partitions where the leader is not on the 
> preferred replica.



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

[jira] [Created] (KAFKA-3355) GetOffsetShell command doesn't work with SASL enabled Kafka

[TAO XIAO (JIRA)]

TAO XIAO created KAFKA-3355:
---

 Summary: GetOffsetShell command doesn't work with SASL enabled 
Kafka
 Key: KAFKA-3355
 URL: https://issues.apache.org/jira/browse/KAFKA-3355
 Project: Kafka
  Issue Type: Bug
  Components: tools
Affects Versions: 0.9.0.1
 Environment: Kafka 0.9.0.1
Reporter: TAO XIAO


I found that GetOffsetShell doesn't work with SASL enabled Kafka. I believe 
this is due to old producer being used in GetOffsetShell.

Kafka version 0.9.0.1

Exception

% bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 
localhost:9092 --topic test --time -1
[2016-03-04 21:43:56,597] INFO Verifying properties 
(kafka.utils.VerifiableProperties)
[2016-03-04 21:43:56,613] INFO Property client.id is overridden to 
GetOffsetShell (kafka.utils.VerifiableProperties)
[2016-03-04 21:43:56,613] INFO Property metadata.broker.list is overridden to 
localhost:9092 (kafka.utils.VerifiableProperties)
[2016-03-04 21:43:56,613] INFO Property request.timeout.ms is overridden to 
1000 (kafka.utils.VerifiableProperties)
[2016-03-04 21:43:56,674] INFO Fetching metadata from broker 
BrokerEndPoint(0,localhost,9092) with correlation id 0 for 1 topic(s) Set(test) 
(kafka.client.ClientUtils$)
[2016-03-04 21:43:56,689] INFO Connected to localhost:9092 for producing 
(kafka.producer.SyncProducer)
[2016-03-04 21:43:56,705] WARN Fetching topic metadata with correlation id 0 
for topics [Set(test)] from broker [BrokerEndPoint(0,localhost,9092)] failed 
(kafka.client.ClientUtils$)
java.nio.BufferUnderflowException
at java.nio.Buffer.nextGetIndex(Buffer.java:498)
at java.nio.HeapByteBuffer.getShort(HeapByteBuffer.java:304)
at kafka.api.ApiUtils$.readShortString(ApiUtils.scala:36)
at kafka.cluster.BrokerEndPoint$.readFrom(BrokerEndPoint.scala:52)
at 
kafka.api.TopicMetadataResponse$$anonfun$1.apply(TopicMetadataResponse.scala:28)
at 
kafka.api.TopicMetadataResponse$$anonfun$1.apply(TopicMetadataResponse.scala:28)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at 
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
at scala.collection.immutable.Range.foreach(Range.scala:166)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at 
kafka.api.TopicMetadataResponse$.readFrom(TopicMetadataResponse.scala:28)
at kafka.producer.SyncProducer.send(SyncProducer.scala:120)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
at kafka.tools.GetOffsetShell$.main(GetOffsetShell.scala:78)
at kafka.tools.GetOffsetShell.main(GetOffsetShell.scala)



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

[jira] [Updated] (KAFKA-3237) ConfigDef validators require a default value

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-3237:
-
Assignee: Jeremy Custenborder

> ConfigDef validators require a default value
> 
>
> Key: KAFKA-3237
> URL: https://issues.apache.org/jira/browse/KAFKA-3237
> Project: Kafka
>  Issue Type: Bug
>  Components: config
>Affects Versions: 0.9.0.0
>Reporter: Jeremy Custenborder
>Assignee: Jeremy Custenborder
>Priority: Minor
> Fix For: 0.10.0.0
>
>
> I should be able to add a ConfigDef that has a validator but does has null as 
> the default value. This would allow me to have a required property that is 
> restricted to certain strings in this example. This exception should be 
> thrown upon call to ConfigDef.parse instead. 
> {code}
> ConfigDef def = new ConfigDef();
> def.define(key, Type.STRING, null, ValidString.in("ONE", "TWO", "THREE"), 
> Importance.HIGH, "docs");
> {code}
> {code}
> Invalid value null for configuration test: String must be one of: ONE, TWO, 
> THREE
> org.apache.kafka.common.config.ConfigException: Invalid value null for 
> configuration enum_test: String must be one of: ONE, TWO, THREE
>   at 
> org.apache.kafka.common.config.ConfigDef$ValidString.ensureValid(ConfigDef.java:349)
>   at 
> org.apache.kafka.common.config.ConfigDef$ConfigKey.(ConfigDef.java:375)
> {code}



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

[jira] [Commented] (KAFKA-3237) ConfigDef validators require a default value

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> ConfigDef validators require a default value
> 
>
> Key: KAFKA-3237
> URL: https://issues.apache.org/jira/browse/KAFKA-3237
> Project: Kafka
>  Issue Type: Bug
>  Components: config
>Affects Versions: 0.9.0.0
>Reporter: Jeremy Custenborder
>Priority: Minor
> Fix For: 0.10.0.0
>
>
> I should be able to add a ConfigDef that has a validator but does has null as 
> the default value. This would allow me to have a required property that is 
> restricted to certain strings in this example. This exception should be 
> thrown upon call to ConfigDef.parse instead. 
> {code}
> ConfigDef def = new ConfigDef();
> def.define(key, Type.STRING, null, ValidString.in("ONE", "TWO", "THREE"), 
> Importance.HIGH, "docs");
> {code}
> {code}
> Invalid value null for configuration test: String must be one of: ONE, TWO, 
> THREE
> org.apache.kafka.common.config.ConfigException: Invalid value null for 
> configuration enum_test: String must be one of: ONE, TWO, THREE
>   at 
> org.apache.kafka.common.config.ConfigDef$ValidString.ensureValid(ConfigDef.java:349)
>   at 
> org.apache.kafka.common.config.ConfigDef$ConfigKey.(ConfigDef.java:375)
> {code}



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

[GitHub] kafka pull request: KAFKA-3237 - Remove test cases testInvalidDefa...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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] [Resolved] (KAFKA-3237) ConfigDef validators require a default value

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava resolved KAFKA-3237.
--
   Resolution: Fixed
Fix Version/s: 0.10.0.0

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

> ConfigDef validators require a default value
> 
>
> Key: KAFKA-3237
> URL: https://issues.apache.org/jira/browse/KAFKA-3237
> Project: Kafka
>  Issue Type: Bug
>  Components: config
>Affects Versions: 0.9.0.0
>Reporter: Jeremy Custenborder
>Priority: Minor
> Fix For: 0.10.0.0
>
>
> I should be able to add a ConfigDef that has a validator but does has null as 
> the default value. This would allow me to have a required property that is 
> restricted to certain strings in this example. This exception should be 
> thrown upon call to ConfigDef.parse instead. 
> {code}
> ConfigDef def = new ConfigDef();
> def.define(key, Type.STRING, null, ValidString.in("ONE", "TWO", "THREE"), 
> Importance.HIGH, "docs");
> {code}
> {code}
> Invalid value null for configuration test: String must be one of: ONE, TWO, 
> THREE
> org.apache.kafka.common.config.ConfigException: Invalid value null for 
> configuration enum_test: String must be one of: ONE, TWO, THREE
>   at 
> org.apache.kafka.common.config.ConfigDef$ValidString.ensureValid(ConfigDef.java:349)
>   at 
> org.apache.kafka.common.config.ConfigDef$ConfigKey.(ConfigDef.java:375)
> {code}



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

[jira] [Updated] (KAFKA-3354) Streams system tests fail in trunk

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3354:
-
Priority: Blocker  (was: Major)

> Streams system tests fail in trunk
> --
>
> Key: KAFKA-3354
> URL: https://issues.apache.org/jira/browse/KAFKA-3354
> Project: Kafka
>  Issue Type: Bug
>Reporter: Geoff Anderson
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> Started seeing consistent failures in 
> {code}
> Module: kafkatest.tests.streams_bounce_test
> Class:  StreamsBounceTest
> Method: test_bounce
> {code}
> and
> {code}
> Module: kafkatest.tests.streams_smoke_test
> Class:  StreamsSmokeTest
> Method: test_streams
> {code}
> Last success:  23f239b
> First starting failing: 2a58ba9
> Two commits between:
> {code}
> commit 2a58ba9fd893979f89aec251579b10f5cda41d10
> Author: Guozhang Wang 
> Date:   Wed Mar 2 13:43:48 2016 -0800
> KAFKA-3311; Prepare internal source topics before calling partition 
> grouper
> Author: Guozhang Wang 
> Reviewers: Yasuhiro Matsuda , Jun Rao 
> 
> Closes #990 from guozhangwang/K3311
> commit 002b377dad9c956cd0ae0597981f29698883b6d5
> Author: Anna Povzner 
> Date:   Wed Mar 2 09:40:34 2016 -0800
> KAFKA-3196; Added checksum and size to RecordMetadata and ConsumerRecord
> This is the second (remaining) part of KIP-42. See 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors
> Author: Anna Povzner 
> Reviewers: Ismael Juma , Jason Gustafson 
> , Jun Rao 
> Closes #951 from apovzner/kafka-3196
> {code}



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

[jira] [Commented] (KAFKA-3354) Streams system tests fail in trunk

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Streams system tests fail in trunk
> --
>
> Key: KAFKA-3354
> URL: https://issues.apache.org/jira/browse/KAFKA-3354
> Project: Kafka
>  Issue Type: Bug
>Reporter: Geoff Anderson
> Fix For: 0.10.0.0
>
>
> Started seeing consistent failures in 
> {code}
> Module: kafkatest.tests.streams_bounce_test
> Class:  StreamsBounceTest
> Method: test_bounce
> {code}
> and
> {code}
> Module: kafkatest.tests.streams_smoke_test
> Class:  StreamsSmokeTest
> Method: test_streams
> {code}
> Last success:  23f239b
> First starting failing: 2a58ba9
> Two commits between:
> {code}
> commit 2a58ba9fd893979f89aec251579b10f5cda41d10
> Author: Guozhang Wang 
> Date:   Wed Mar 2 13:43:48 2016 -0800
> KAFKA-3311; Prepare internal source topics before calling partition 
> grouper
> Author: Guozhang Wang 
> Reviewers: Yasuhiro Matsuda , Jun Rao 
> 
> Closes #990 from guozhangwang/K3311
> commit 002b377dad9c956cd0ae0597981f29698883b6d5
> Author: Anna Povzner 
> Date:   Wed Mar 2 09:40:34 2016 -0800
> KAFKA-3196; Added checksum and size to RecordMetadata and ConsumerRecord
> This is the second (remaining) part of KIP-42. See 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors
> Author: Anna Povzner 
> Reviewers: Ismael Juma , Jason Gustafson 
> , Jun Rao 
> Closes #951 from apovzner/kafka-3196
> {code}



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

[jira] [Updated] (KAFKA-3354) Streams system tests fail in trunk

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3354:
-
Fix Version/s: 0.10.0.0

> Streams system tests fail in trunk
> --
>
> Key: KAFKA-3354
> URL: https://issues.apache.org/jira/browse/KAFKA-3354
> Project: Kafka
>  Issue Type: Bug
>Reporter: Geoff Anderson
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> Started seeing consistent failures in 
> {code}
> Module: kafkatest.tests.streams_bounce_test
> Class:  StreamsBounceTest
> Method: test_bounce
> {code}
> and
> {code}
> Module: kafkatest.tests.streams_smoke_test
> Class:  StreamsSmokeTest
> Method: test_streams
> {code}
> Last success:  23f239b
> First starting failing: 2a58ba9
> Two commits between:
> {code}
> commit 2a58ba9fd893979f89aec251579b10f5cda41d10
> Author: Guozhang Wang 
> Date:   Wed Mar 2 13:43:48 2016 -0800
> KAFKA-3311; Prepare internal source topics before calling partition 
> grouper
> Author: Guozhang Wang 
> Reviewers: Yasuhiro Matsuda , Jun Rao 
> 
> Closes #990 from guozhangwang/K3311
> commit 002b377dad9c956cd0ae0597981f29698883b6d5
> Author: Anna Povzner 
> Date:   Wed Mar 2 09:40:34 2016 -0800
> KAFKA-3196; Added checksum and size to RecordMetadata and ConsumerRecord
> This is the second (remaining) part of KIP-42. See 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors
> Author: Anna Povzner 
> Reviewers: Ismael Juma , Jason Gustafson 
> , Jun Rao 
> Closes #951 from apovzner/kafka-3196
> {code}



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

[GitHub] kafka pull request: MINOR: Ignoring streams tests until there is f...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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.
---

Re: Additional Kafka Connect schema logical types?

[Ewen Cheslack-Postava]

On Fri, Mar 4, 2016 at 1:37 PM, Randall Hauch  wrote:

> I’m working on a Kafka Connect connector that reads a MySQL binlog to
> provide near real-time change data capture, and I also plan connectors for
> other DBMSes. The problem is that I’m not able to map all of the MySQL data
> types — or even all of the standard JDBC types — to Kafka Connect Schemas
> without resorting to complex Schemas that radically increase the footprint
> of messages.
>
> Specifically, I’d like my connectors to be able to use the following
> “logical” types:
>
> - Bits: A set of bits of arbitrary length, corresponding to
> java.util.BitSet. See [1] for code..
> - IsoTime: An ISO8601 time that includes the time zone and corresponding
> to Java 8’s java.time.OffsetTime that represents a time with the offset
> from UTC/Greenwich, and that has a well-defined ordering and thus is more
> suitable for persistent storage. See [2] for code..
> - IsoTimestamp: An ISO8601 timestamp that includes the time zone and
> corresponding to Java 8’s java.time.OffsetDateTime that represents an
> instant with the offset from UTC/Greenwich, and that has a well-defined
> ordering and thus is more suitable for persistent storage. See [3] for code.
>

These all makes sense and seem pretty straightforward. I'm hoping to scope
the available logical types in the framework so they cover as many cases as
possible with as few types as possible, precisely for the reason you
mention below -- the number of types can quickly explode, which is bad for
everyone.

(And a note probably better left to review if you submit these as new
logical types for the framework: for BitSet, it doesn't seem like the
encoding preserves the number of bits, i.e. it looks like it might round up
to 8. Should the number of bits be a schema parameter, or is the loss ok
and we can't guarantee fixed sizes for common use cases?)


> These are very similar to the 4 built-in logical types (Decimal, Date,
> Time, and Timestamp). These logical types are much akin to aliases for a
> primitive type (typically BYTES), and their use within a Schema includes
> semantics that would not be there by just using the corresponding primitive.
>
> Unfortunately, Kafka Connect is not currently able to support custom
> logical types. Sure, you can create them, since the JsonConverter (nor any
> of the other Converters) will know how to serialize or deserialize them.
>
> One option is for Kafka Connect to add these, but this is sort of a
> never-ending battle. And, since Kafka is not yet on Java 8, supporting
> OffsetTime and OffsetDateTime would be problematic.
>
> Perhaps a better option is to support custom logical types, where each
> logical type must be based upon a single primitive type and must define a
> class that knows how to serialize and deserialize the logical type from the
> primitive type. The Converters, once modified, could look for the
> referenced class and use its serdes logic as needed.
>

Possibly simpler would be to make them pluggable in the framework but not
require the Converters to be aware of them. This requires care in the
framework to ensure we handle to/fromLogical conversions everywhere a
conversion takes place, but avoids complicating Converters. Even better is
if we can do this in a backwards compatible way such that even if the
Converter does the logical conversions (as they currently do) that check
just becomes a nop if the conversion has already been performed.

However, not baking logical types into the framework makes it way more
complicated to use them. What's the registration process? Do we need to
pick up the classes automatically somehow? One of the nice things about
baking them into the framework is that it helps define the set of valid
Java types that can be passed to converters. With a completely pluggable
solution, Converters can't be sure what they're allowed to do/what types
they need to support.


>
> A couple of points:
>
> 1) Any source connector that is producing a record with these logical
> types would obviously have to have the logical type’s class available on
> the classpath. That doesn’t seem a difficult requirement to satisfy.
>

This seems easy, but keep in mind how this can complicate compatibility
when you have multiple connector plugins depending on the same jar for
logical types but on different versions. This is of course always a problem
for dependencies, but structuring logical types like this seems like it'll
very quickly cause us to hit these problems. At a minimum, it makes
compatibility harder to reason about.


>
> 2) Any consumer or source connector that is consuming records with these
> values needs to be able to work with the logical type’s class to be able to
> work with it. This doesn’t seem too horrible, especially if the logical
> type class(es) are nicely separated into separate JARs. However, if the
> consumer doesn’t have the logical type class, then its local Converter
> would just deserialize to the 

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

[Apache Jenkins Server]

See 

Changes:

[junrao] KAFKA-3343; Use NoTimestamp in GroupMetadataManager when message v0 
i…

--
[...truncated 5802 lines...]
org.apache.kafka.connect.runtime.WorkerTest > testAddConnectorByAlias PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddConnectorByShortAlias 
PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStopInvalidConnector PASSED

org.apache.kafka.connect.runtime.WorkerTest > testReconfigureConnectorTasks 
PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddRemoveTask PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStopInvalidTask PASSED

org.apache.kafka.connect.runtime.WorkerTest > testCleanupTasksOnStop PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > testPollRedelivery PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > 
testErrorInRebalancePartitionRevocation PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > 
testErrorInRebalancePartitionAssignment PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > stopBeforeStarting PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > standardStartup PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > cancelBeforeStopping PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testPollsInBackground 
PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testFailureInPoll PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testCommit PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testCommitFailure PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > 
testSendRecordsConvertsData PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testSendRecordsRetries 
PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testSlowTaskStart PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testCreateSinkConnector PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testDestroyConnector PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testCreateAndStop PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testAccessors PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testCreateConnectorAlreadyExists PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testCreateSourceConnector PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testPutConnectorConfig PASSED

org.apache.kafka.connect.runtime.standalone.StandaloneHerderTest > 
testPutTaskConfigs PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testJoinAssignment PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testRebalance PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testHaltCleansUpWorker PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testCreateConnector PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testConnectorConfigAdded PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testConnectorConfigUpdate PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testTaskConfigAdded PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testJoinLeaderCatchUpFails PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testInconsistentConfigs PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testDestroyConnector PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testAccessors PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testCreateConnectorAlreadyExists PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testPutConnectorConfig PASSED

org.apache.kafka.connect.runtime.distributed.WorkerCoordinatorTest > 
testNormalJoinGroupFollower PASSED

org.apache.kafka.connect.runtime.distributed.WorkerCoordinatorTest > 
testMetadata PASSED

org.apache.kafka.connect.runtime.distributed.WorkerCoordinatorTest > 
testLeaderPerformAssignment1 PASSED

org.apache.kafka.connect.runtime.distributed.WorkerCoordinatorTest > 
testLeaderPerformAssignment2 PASSED

org.apache.kafka.connect.runtime.distributed.WorkerCoordinatorTest > 
testJoinLeaderCannotAssign PASSED

org.apache.kafka.connect.runtime.distributed.WorkerCoordinatorTest > 
testRejoinGroup PASSED

org.apache.kafka.connect.runtime.distributed.WorkerCoordinatorTest > 
testNormalJoinGroupLeader PASSED

org.apache.kafka.connect.runtime.AbstractHerderTest > connectorStatus PASSED

org.apache.kafka.connect.runtime.AbstractHerderTest > taskStatus PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskThreadedTest > 
testPollsInBackground PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskThreadedTest > 

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

[Apache Jenkins Server]

See 

[GitHub] kafka pull request: MINOR: Ignoring streams tests until there is f...

[granders]

GitHub user granders opened a pull request:

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

MINOR: Ignoring streams tests until there is fix for KAFKA-3354

Per discussion with @guozhangwang, `@ignore` failing streams system tests 
until fix for KAFKA-3354 is checked in.

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

$ git pull https://github.com/confluentinc/kafka ignore-streams-systest

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

https://github.com/apache/kafka/pull/1031.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 #1031


commit 90b72c5fecd3b5646ce8d3f24ba5522023dacc96
Author: Geoff Anderson 
Date:   2016-03-09T02:36:26Z

Ignoring streams tests until there is fix for KAFKA-3354




---
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-3354) Streams system tests fail in trunk

[ASF GitHub Bot (JIRA)]

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

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

GitHub user granders opened a pull request:

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

MINOR: Ignoring streams tests until there is fix for KAFKA-3354

Per discussion with @guozhangwang, `@ignore` failing streams system tests 
until fix for KAFKA-3354 is checked in.

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

$ git pull https://github.com/confluentinc/kafka ignore-streams-systest

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

https://github.com/apache/kafka/pull/1031.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 #1031


commit 90b72c5fecd3b5646ce8d3f24ba5522023dacc96
Author: Geoff Anderson 
Date:   2016-03-09T02:36:26Z

Ignoring streams tests until there is fix for KAFKA-3354




> Streams system tests fail in trunk
> --
>
> Key: KAFKA-3354
> URL: https://issues.apache.org/jira/browse/KAFKA-3354
> Project: Kafka
>  Issue Type: Bug
>Reporter: Geoff Anderson
>
> Started seeing consistent failures in 
> {code}
> Module: kafkatest.tests.streams_bounce_test
> Class:  StreamsBounceTest
> Method: test_bounce
> {code}
> and
> {code}
> Module: kafkatest.tests.streams_smoke_test
> Class:  StreamsSmokeTest
> Method: test_streams
> {code}
> Last success:  23f239b
> First starting failing: 2a58ba9
> Two commits between:
> {code}
> commit 2a58ba9fd893979f89aec251579b10f5cda41d10
> Author: Guozhang Wang 
> Date:   Wed Mar 2 13:43:48 2016 -0800
> KAFKA-3311; Prepare internal source topics before calling partition 
> grouper
> Author: Guozhang Wang 
> Reviewers: Yasuhiro Matsuda , Jun Rao 
> 
> Closes #990 from guozhangwang/K3311
> commit 002b377dad9c956cd0ae0597981f29698883b6d5
> Author: Anna Povzner 
> Date:   Wed Mar 2 09:40:34 2016 -0800
> KAFKA-3196; Added checksum and size to RecordMetadata and ConsumerRecord
> This is the second (remaining) part of KIP-42. See 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors
> Author: Anna Povzner 
> Reviewers: Ismael Juma , Jason Gustafson 
> , Jun Rao 
> Closes #951 from apovzner/kafka-3196
> {code}



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

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

[Apache Jenkins Server]

See 

Changes:

[jjkoshy] KAFKA-3197; Fix producer sending records out of order

--
[...truncated 3568 lines...]

org.apache.kafka.clients.ClientUtilsTest > testNoPort PASSED

org.apache.kafka.clients.NetworkClientTest > testSimpleRequestResponse PASSED

org.apache.kafka.clients.NetworkClientTest > testClose PASSED

org.apache.kafka.clients.NetworkClientTest > testLeastLoadedNode PASSED

org.apache.kafka.clients.NetworkClientTest > testConnectionDelayConnected PASSED

org.apache.kafka.clients.NetworkClientTest > testRequestTimeout PASSED

org.apache.kafka.clients.NetworkClientTest > 
testSimpleRequestResponseWithStaticNodes PASSED

org.apache.kafka.clients.NetworkClientTest > testConnectionDelay PASSED

org.apache.kafka.clients.NetworkClientTest > testSendToUnreadyNode PASSED

org.apache.kafka.clients.NetworkClientTest > testConnectionDelayDisconnected 
PASSED

org.apache.kafka.clients.MetadataTest > testListenerCanUnregister PASSED

org.apache.kafka.clients.MetadataTest > testFailedUpdate PASSED

org.apache.kafka.clients.MetadataTest > testMetadataUpdateWaitTime PASSED

org.apache.kafka.clients.MetadataTest > testUpdateWithNeedMetadataForAllTopics 
PASSED

org.apache.kafka.clients.MetadataTest > testMetadata PASSED

org.apache.kafka.clients.MetadataTest > testListenerGetsNotifiedOfUpdate PASSED

org.apache.kafka.common.SerializeCompatibilityTopicPartitionTest > 
testSerializationRoundtrip PASSED

org.apache.kafka.common.SerializeCompatibilityTopicPartitionTest > 
testTopiPartitionSerializationCompatibility PASSED

org.apache.kafka.common.serialization.SerializationTest > testStringSerializer 
PASSED

org.apache.kafka.common.serialization.SerializationTest > testIntegerSerializer 
PASSED

org.apache.kafka.common.serialization.SerializationTest > 
testByteBufferSerializer PASSED

org.apache.kafka.common.config.AbstractConfigTest > testOriginalsWithPrefix 
PASSED

org.apache.kafka.common.config.AbstractConfigTest > testConfiguredInstances 
PASSED

org.apache.kafka.common.config.ConfigDefTest > testBasicTypes PASSED

org.apache.kafka.common.config.ConfigDefTest > testNullDefault PASSED

org.apache.kafka.common.config.ConfigDefTest > testInvalidDefaultRange PASSED

org.apache.kafka.common.config.ConfigDefTest > testValidators PASSED

org.apache.kafka.common.config.ConfigDefTest > testInvalidDefaultString PASSED

org.apache.kafka.common.config.ConfigDefTest > testSslPasswords PASSED

org.apache.kafka.common.config.ConfigDefTest > testInvalidDefault PASSED

org.apache.kafka.common.config.ConfigDefTest > testMissingRequired PASSED

org.apache.kafka.common.config.ConfigDefTest > testDefinedTwice PASSED

org.apache.kafka.common.config.ConfigDefTest > testBadInputs PASSED

org.apache.kafka.common.protocol.ErrorsTest > testForExceptionDefault PASSED

org.apache.kafka.common.protocol.ErrorsTest > testUniqueExceptions PASSED

org.apache.kafka.common.protocol.ErrorsTest > testForExceptionInheritance PASSED

org.apache.kafka.common.protocol.ErrorsTest > testNoneException PASSED

org.apache.kafka.common.protocol.ErrorsTest > testUniqueErrorCodes PASSED

org.apache.kafka.common.protocol.ErrorsTest > testExceptionsAreNotGeneric PASSED

org.apache.kafka.common.protocol.ApiKeysTest > testForIdWithInvalidIdLow PASSED

org.apache.kafka.common.protocol.ApiKeysTest > testForIdWithInvalidIdHigh PASSED

org.apache.kafka.common.protocol.ProtoUtilsTest > schemaVersionOutOfRange PASSED

org.apache.kafka.common.protocol.types.ProtocolSerializationTest > testArray 
PASSED

org.apache.kafka.common.protocol.types.ProtocolSerializationTest > testNulls 
PASSED

org.apache.kafka.common.protocol.types.ProtocolSerializationTest > 
testNullableDefault PASSED

org.apache.kafka.common.protocol.types.ProtocolSerializationTest > testDefault 
PASSED

org.apache.kafka.common.protocol.types.ProtocolSerializationTest > testSimple 
PASSED

org.apache.kafka.common.requests.RequestResponseTest > testSerialization PASSED

org.apache.kafka.common.requests.RequestResponseTest > fetchResponseVersionTest 
PASSED

org.apache.kafka.common.requests.RequestResponseTest > 
produceResponseVersionTest PASSED

org.apache.kafka.common.requests.RequestResponseTest > 
testControlledShutdownResponse PASSED

org.apache.kafka.common.requests.RequestResponseTest > 
testRequestHeaderWithNullClientId PASSED

org.apache.kafka.common.security.auth.KafkaPrincipalTest > 
testPrincipalNameCanContainSeparator PASSED

org.apache.kafka.common.security.auth.KafkaPrincipalTest > 
testEqualsAndHashCode PASSED

org.apache.kafka.common.security.kerberos.KerberosNameTest > testParse PASSED

org.apache.kafka.common.security.ssl.SslFactoryTest > testClientMode PASSED

org.apache.kafka.common.security.ssl.SslFactoryTest > 
testSslFactoryConfiguration PASSED

org.apache.kafka.common.metrics.MetricsTest > testSimpleStats PASSED

org.apache.kafka.common.metrics.MetricsTest > 

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

[Apache Jenkins Server]

See 

Changes:

[jjkoshy] KAFKA-3197; Fix producer sending records out of order

--
[...truncated 2971 lines...]

kafka.integration.SaslSslTopicMetadataTest > testGetAllTopicMetadata PASSED

kafka.integration.SaslSslTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup PASSED

kafka.integration.SaslSslTopicMetadataTest > testBasicTopicMetadata PASSED

kafka.integration.SaslSslTopicMetadataTest > testTopicMetadataRequest PASSED

kafka.integration.SaslSslTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown PASSED

kafka.integration.SslTopicMetadataTest > testIsrAfterBrokerShutDownAndJoinsBack 
PASSED

kafka.integration.SslTopicMetadataTest > testAutoCreateTopicWithCollision PASSED

kafka.integration.SslTopicMetadataTest > testAliveBrokerListWithNoTopics PASSED

kafka.integration.SslTopicMetadataTest > testAutoCreateTopic PASSED

kafka.integration.SslTopicMetadataTest > testGetAllTopicMetadata PASSED

kafka.integration.SslTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup PASSED

kafka.integration.SslTopicMetadataTest > testBasicTopicMetadata PASSED

kafka.integration.SslTopicMetadataTest > testTopicMetadataRequest PASSED

kafka.integration.SslTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown PASSED

kafka.integration.AutoOffsetResetTest > testResetToLatestWhenOffsetTooLow PASSED

kafka.integration.AutoOffsetResetTest > testResetToEarliestWhenOffsetTooLow 
PASSED

kafka.integration.AutoOffsetResetTest > testResetToEarliestWhenOffsetTooHigh 
PASSED

kafka.integration.AutoOffsetResetTest > testResetToLatestWhenOffsetTooHigh 
PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testIsrAfterBrokerShutDownAndJoinsBack PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAutoCreateTopicWithCollision PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokerListWithNoTopics PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > testAutoCreateTopic PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > testGetAllTopicMetadata 
PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > testBasicTopicMetadata PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > testTopicMetadataRequest 
PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown PASSED

kafka.KafkaTest > testGetKafkaConfigFromArgsWrongSetValue PASSED

kafka.KafkaTest > testKafkaSslPasswords PASSED

kafka.KafkaTest > testGetKafkaConfigFromArgs PASSED

kafka.KafkaTest > testGetKafkaConfigFromArgsNonArgsAtTheEnd PASSED

kafka.KafkaTest > testGetKafkaConfigFromArgsNonArgsOnly PASSED

kafka.KafkaTest > testGetKafkaConfigFromArgsNonArgsAtTheBegging PASSED

kafka.metrics.KafkaTimerTest > testKafkaTimer PASSED

kafka.utils.UtilsTest > testAbs PASSED

kafka.utils.UtilsTest > testReplaceSuffix PASSED

kafka.utils.UtilsTest > testCircularIterator PASSED

kafka.utils.UtilsTest > testReadBytes PASSED

kafka.utils.UtilsTest > testCsvList PASSED

kafka.utils.UtilsTest > testReadInt PASSED

kafka.utils.UtilsTest > testCsvMap PASSED

kafka.utils.UtilsTest > testInLock PASSED

kafka.utils.UtilsTest > testSwallow PASSED

kafka.utils.SchedulerTest > testMockSchedulerNonPeriodicTask PASSED

kafka.utils.SchedulerTest > testMockSchedulerPeriodicTask PASSED

kafka.utils.SchedulerTest > testNonPeriodicTask PASSED

kafka.utils.SchedulerTest > testRestart PASSED

kafka.utils.SchedulerTest > testReentrantTaskInMockScheduler PASSED

kafka.utils.SchedulerTest > testPeriodicTask PASSED

kafka.utils.ByteBoundedBlockingQueueTest > testByteBoundedBlockingQueue PASSED

kafka.utils.timer.TimerTaskListTest > testAll PASSED

kafka.utils.timer.TimerTest > testAlreadyExpiredTask PASSED

kafka.utils.timer.TimerTest > testTaskExpiration PASSED

kafka.utils.CommandLineUtilsTest > testParseEmptyArg PASSED

kafka.utils.CommandLineUtilsTest > testParseSingleArg PASSED

kafka.utils.CommandLineUtilsTest > testParseArgs PASSED

kafka.utils.CommandLineUtilsTest > testParseEmptyArgAsValid PASSED

kafka.utils.IteratorTemplateTest > testIterator PASSED

kafka.utils.ReplicationUtilsTest > testUpdateLeaderAndIsr PASSED

kafka.utils.ReplicationUtilsTest > testGetLeaderIsrAndEpochForPartition PASSED

kafka.utils.JsonTest > testJsonEncoding PASSED

kafka.message.MessageCompressionTest > testCompressSize PASSED

kafka.message.MessageCompressionTest > testSimpleCompressDecompress PASSED

kafka.message.MessageWriterTest > testWithNoCompressionAttribute PASSED

kafka.message.MessageWriterTest > testWithCompressionAttribute PASSED

kafka.message.MessageWriterTest > testBufferingOutputStream PASSED

kafka.message.MessageWriterTest > testWithKey PASSED

kafka.message.MessageTest > testChecksum PASSED

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Gwen Shapira]

"Since smart card logins are not built into Kafka (or the JDK), you need a
developer to build the login module. So the developer implements
example.SmartcardLoginModule. In addition, the developer may also implement
callback handlers for the SASL client or server  and a login class to keep
this login refreshed. The callback handlers and login implementation
support all the mechanisms that the organisation supports - in this case
Kerberos and smart card."

In this case, the developer works for Example Inc (which develops
SmartCard authentication modules), while I work for Retailer and need
to use his module.
You assume that developer from Example Inc knows about all the
mechanisms that I have enabled in my Kafka cluster and he is capable
of developing a Login class and maybe a callback handler that supports
all of them. I think this is an unreasonable expectation.

I think that allowing plug in of Login method and single Callback
handler is not a good extensible solution for multiple mechanisms. We
should probably supply a Kafka-based LoginContext and CallbackHandler
that will be able to delegate to implementations of LoginModules and
Callback Handlers based on mechanism out of a list supplied by a user.
I agree it is more complex, but otherwise we are asking of too much
coordination from whoever implements CallbackHandlers and Login
methods.

Gwen

On Tue, Mar 8, 2016 at 3:35 PM, Rajini Sivaram
 wrote:
> As an admin, you would configure login modules using standard JAAS
> configuration. So, if you are supporting Kerberos and smart card login
> modules within a single broker, in your jaas.conf you would specify:
>
> KafkaServer {
>
> com.ibm.security.auth.module.Krb5LoginModule required
> credsType=both
> useKeytab="file:/kafka/key.tab"
> principal="kafka/localh...@example.com";
>
>   example.SmartcardLoginModule required
>
>   cardNumber=123
>
> };
>
>
> Since smart card logins are not built into Kafka (or the JDK), you need a
> developer to build the login module. So the developer implements
> example.SmartcardLoginModule. In addition, the developer may also implement
> callback handlers for the SASL client or server  and a login class to keep
> this login refreshed. The callback handlers and login implementation
> support all the mechanisms that the organisation supports - in this case
> Kerberos and smart card.
>
> As an admin, if you are using mechanisms built into Kafka, you simply
> specify a JAAS config with the combination of mechanisms you want and you
> are done. If you are adding new mechanisms, it clearly can't be done
> without some additional code to implement that mechanism. The callback and
> login interfaces are configurable to ensure that (as a developer), you can
> build custom mechanisms without modifying Kafka. SASL was designed
> precisely to do this. But at the moment, Kafka restricts what type of
> authentication you can perform with SASL.
>
> Back to the questions:
> 1. *How does the proposed Login class interact with LoginContext and *
> *LoginModule.*
> Login class performs *LoginContext.login(). *This performs login for all
> the login modules of that context (this is standard Java function). Login
> is cached with a reference counter to keep it alive.
>
> 2. *Do we really need a pluggable Login class, when users **(administrators,
> not developers) are more likely to want to specify LoginModules?*
> Administrators specify *LoginModules. *Developers implement new mechanisms.
> The pluggable *Login* class enables the developer to implement new
> mechanisms without changing Kafka. If Kafka didn't have a default Kerberos
> implementation and a user wanted to use Kerberos, pluggability of *Login* 
> class
> enables the user to include Kerberos without changing Kafka. Not that we
> want another Kerberos, but Tao had a requirement to enable a custom
> mechanism with functionality similar to Kerberos (see KIP-44). Pluggable
> *Login* supports this use case without replacing the entire authentication
> mechanism in Kafka.
>
> *3. How will users get to specify multiple LoginModules?*
> Standard Java JAAS configuration - see example above.
>
>
> *4. It sounds like different LoginModules may need
> different CallbackHandlers. Since we only allow a single
> pluggableCallbackHandler class in the KIP-43 configuration, how will this
> be handled?*
> One LoginContext, one subject, one callback handler, multiple login
> modules. This is the Java model for login. And it works well because
> callback handlers tend to be fairly standard. Even if the smart card SASL
> callbacks needed to do something wildly different and you wanted a broker
> that combined this with Kerberos, you would implement a callback that
> looked like the code below.
>
> private CallbackHandler handler;
>
> public void configure(Map configs, Mode mode, Subject
> subject, String mechanism) {public void configure(Map configs,
> Mode mode, Subject subject, 

[jira] [Updated] (KAFKA-3353) Remove deprecated producer configs.

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3353:
--
Description: 
Following producer configs were deprecated in 0.9, it will be a good idea to 
remove them in 0.11. Removing in 0.10 was not an option, as 0.10 was a short 
release and direct upgrades from 0.8 to 0.10 are supported.

* block.on.buffer.full
* metadata.fetch.timeout.ms
* timeout.ms

  was:
Following producer configs were deprecated in 0.9, it will be a good idea to 
remove them in 0.11.

* block.on.buffer.full
* metadata.fetch.timeout.ms
* timeout.ms


> Remove deprecated producer configs.
> ---
>
> Key: KAFKA-3353
> URL: https://issues.apache.org/jira/browse/KAFKA-3353
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
>
> Following producer configs were deprecated in 0.9, it will be a good idea to 
> remove them in 0.11. Removing in 0.10 was not an option, as 0.10 was a short 
> release and direct upgrades from 0.8 to 0.10 are supported.
> * block.on.buffer.full
> * metadata.fetch.timeout.ms
> * timeout.ms



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

[jira] [Updated] (KAFKA-3353) Remove deprecated producer configs.

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3353:
--
Affects Version/s: 0.11.0.0

> Remove deprecated producer configs.
> ---
>
> Key: KAFKA-3353
> URL: https://issues.apache.org/jira/browse/KAFKA-3353
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
>
> Following producer configs were deprecated in 0.9, it will be a good idea to 
> remove them in 0.11.
> * block.on.buffer.full
> * metadata.fetch.timeout.ms
> * timeout.ms



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

[jira] [Updated] (KAFKA-3353) Remove deprecated producer configs.

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3353:
--
Description: 
Following producer configs were deprecated in 0.9, it will be a good idea to 
remove them in 0.11.

* block.on.buffer.full
* metadata.fetch.timeout.ms
* timeout.ms

  was:
Following producer configs were deprecated in 0.9, it will be a good idea to 
remove them in 0.10.

* block.on.buffer.full
* metadata.fetch.timeout.ms
* timeout.ms


> Remove deprecated producer configs.
> ---
>
> Key: KAFKA-3353
> URL: https://issues.apache.org/jira/browse/KAFKA-3353
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
>
> Following producer configs were deprecated in 0.9, it will be a good idea to 
> remove them in 0.11.
> * block.on.buffer.full
> * metadata.fetch.timeout.ms
> * timeout.ms



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

Re: [Discuss] Remove producer configs deprecated in 0.9

[Ashish Singh]

OK, in that case I will move the JIRA to target for 0.11 instead.

On Tue, Mar 8, 2016 at 4:26 PM, Gwen Shapira  wrote:

> We are planning on supporting direct 0.8.2 to 0.10 upgrades (and are
> adding tests to that effect). So we expect that quite a few users will
> never see 0.9 and therefore never see the deprecation notice.
>
> I don't think this allows us to remove them at this release.
>
> On Tue, Mar 8, 2016 at 3:59 PM, Ashish Singh  wrote:
> > Hey Guys,
> >
> > Following producer configs were deprecated in 0.9, it might be a good
> idea
> > to remove them in 0.10.
> >
> >- block.on.buffer.full
> >- metadata.fetch.timeout.ms
> >- timeout.ms
> >
> >
> > Does this sound OK? KAFKA-3353 is tracking this.
> >
> > --
> >
> > Regards,
> > Ashish
>



-- 

Regards,
Ashish

Re: [Discuss] Remove producer configs deprecated in 0.9

[Ismael Juma]

Thanks Gwen, I was going to say the exact same thing. :)

Ismael

On Wed, Mar 9, 2016 at 12:26 AM, Gwen Shapira  wrote:

> We are planning on supporting direct 0.8.2 to 0.10 upgrades (and are
> adding tests to that effect). So we expect that quite a few users will
> never see 0.9 and therefore never see the deprecation notice.
>
> I don't think this allows us to remove them at this release.
>
> On Tue, Mar 8, 2016 at 3:59 PM, Ashish Singh  wrote:
> > Hey Guys,
> >
> > Following producer configs were deprecated in 0.9, it might be a good
> idea
> > to remove them in 0.10.
> >
> >- block.on.buffer.full
> >- metadata.fetch.timeout.ms
> >- timeout.ms
> >
> >
> > Does this sound OK? KAFKA-3353 is tracking this.
> >
> > --
> >
> > Regards,
> > Ashish
>

[jira] [Updated] (KAFKA-3343) GroupMetadataManager should use NoTimestamp for message v0

[Jun Rao (JIRA)]

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

Jun Rao updated KAFKA-3343:
---
Resolution: Fixed
Status: Resolved  (was: Patch Available)

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

> GroupMetadataManager should use NoTimestamp for message v0
> --
>
> Key: KAFKA-3343
> URL: https://issues.apache.org/jira/browse/KAFKA-3343
> Project: Kafka
>  Issue Type: Bug
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> When message version for __consumer_offsets is v0, the timestamp should be 
> NoTimestamp. Otherwise message instantiation will fail.



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

[jira] [Commented] (KAFKA-3343) GroupMetadataManager should use NoTimestamp for message v0

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> GroupMetadataManager should use NoTimestamp for message v0
> --
>
> Key: KAFKA-3343
> URL: https://issues.apache.org/jira/browse/KAFKA-3343
> Project: Kafka
>  Issue Type: Bug
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> When message version for __consumer_offsets is v0, the timestamp should be 
> NoTimestamp. Otherwise message instantiation will fail.



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

[GitHub] kafka pull request: KAFKA-3343: Use NoTimestamp in GroupMetadataMa...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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.
---

Re: [Discuss] Remove producer configs deprecated in 0.9

[Gwen Shapira]

We are planning on supporting direct 0.8.2 to 0.10 upgrades (and are
adding tests to that effect). So we expect that quite a few users will
never see 0.9 and therefore never see the deprecation notice.

I don't think this allows us to remove them at this release.

On Tue, Mar 8, 2016 at 3:59 PM, Ashish Singh  wrote:
> Hey Guys,
>
> Following producer configs were deprecated in 0.9, it might be a good idea
> to remove them in 0.10.
>
>- block.on.buffer.full
>- metadata.fetch.timeout.ms
>- timeout.ms
>
>
> Does this sound OK? KAFKA-3353 is tracking this.
>
> --
>
> Regards,
> Ashish

[jira] [Created] (KAFKA-3354) Streams system tests fail in trunk

[Geoff Anderson (JIRA)]

Geoff Anderson created KAFKA-3354:
-

 Summary: Streams system tests fail in trunk
 Key: KAFKA-3354
 URL: https://issues.apache.org/jira/browse/KAFKA-3354
 Project: Kafka
  Issue Type: Bug
Reporter: Geoff Anderson


Started seeing consistent failures in 

{code}
Module: kafkatest.tests.streams_bounce_test
Class:  StreamsBounceTest
Method: test_bounce
{code}
and
{code}
Module: kafkatest.tests.streams_smoke_test
Class:  StreamsSmokeTest
Method: test_streams
{code}

Last success:  23f239b
First starting failing: 2a58ba9

Two commits between:

{code}
commit 2a58ba9fd893979f89aec251579b10f5cda41d10
Author: Guozhang Wang 
Date:   Wed Mar 2 13:43:48 2016 -0800

KAFKA-3311; Prepare internal source topics before calling partition grouper

Author: Guozhang Wang 

Reviewers: Yasuhiro Matsuda , Jun Rao 


Closes #990 from guozhangwang/K3311

commit 002b377dad9c956cd0ae0597981f29698883b6d5
Author: Anna Povzner 
Date:   Wed Mar 2 09:40:34 2016 -0800

KAFKA-3196; Added checksum and size to RecordMetadata and ConsumerRecord

This is the second (remaining) part of KIP-42. See 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors

Author: Anna Povzner 

Reviewers: Ismael Juma , Jason Gustafson 
, Jun Rao 

Closes #951 from apovzner/kafka-3196
{code}



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

[jira] [Commented] (KAFKA-3353) Remove deprecated producer configs.

[Ashish K Singh (JIRA)]

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

Ashish K Singh commented on KAFKA-3353:
---

Sent a mail to dev list just a few seconds back :). If there are no major 
objections, removing this will avoid a bit of confusing state of producer 
configs.

> Remove deprecated producer configs.
> ---
>
> Key: KAFKA-3353
> URL: https://issues.apache.org/jira/browse/KAFKA-3353
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
>
> Following producer configs were deprecated in 0.9, it will be a good idea to 
> remove them in 0.10.
> * block.on.buffer.full
> * metadata.fetch.timeout.ms
> * timeout.ms



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

[Discuss] Remove producer configs deprecated in 0.9

[Ashish Singh]

Hey Guys,

Following producer configs were deprecated in 0.9, it might be a good idea
to remove them in 0.10.

   - block.on.buffer.full
   - metadata.fetch.timeout.ms
   - timeout.ms


Does this sound OK? KAFKA-3353 is tracking this.

-- 

Regards,
Ashish

[jira] [Commented] (KAFKA-3353) Remove deprecated producer configs.

[Ismael Juma (JIRA)]

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

Ismael Juma commented on KAFKA-3353:


Given the short release cycle, I think it may make sense to remove them in the 
release after 0.10.0, but not sure what others think.

> Remove deprecated producer configs.
> ---
>
> Key: KAFKA-3353
> URL: https://issues.apache.org/jira/browse/KAFKA-3353
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
>
> Following producer configs were deprecated in 0.9, it will be a good idea to 
> remove them in 0.10.
> * block.on.buffer.full
> * metadata.fetch.timeout.ms
> * timeout.ms



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

[jira] [Created] (KAFKA-3353) Remove deprecated producer configs.

[Ashish K Singh (JIRA)]

Ashish K Singh created KAFKA-3353:
-

 Summary: Remove deprecated producer configs.
 Key: KAFKA-3353
 URL: https://issues.apache.org/jira/browse/KAFKA-3353
 Project: Kafka
  Issue Type: Bug
Reporter: Ashish K Singh
Assignee: Ashish K Singh


Following producer configs were deprecated in 0.9, it will be a good idea to 
remove them in 0.10.

* block.on.buffer.full
* metadata.fetch.timeout.ms
* timeout.ms



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

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Rajini Sivaram]

As an admin, you would configure login modules using standard JAAS
configuration. So, if you are supporting Kerberos and smart card login
modules within a single broker, in your jaas.conf you would specify:

KafkaServer {

com.ibm.security.auth.module.Krb5LoginModule required
credsType=both
useKeytab="file:/kafka/key.tab"
principal="kafka/localh...@example.com";

  example.SmartcardLoginModule required

  cardNumber=123

};


Since smart card logins are not built into Kafka (or the JDK), you need a
developer to build the login module. So the developer implements
example.SmartcardLoginModule. In addition, the developer may also implement
callback handlers for the SASL client or server  and a login class to keep
this login refreshed. The callback handlers and login implementation
support all the mechanisms that the organisation supports - in this case
Kerberos and smart card.

As an admin, if you are using mechanisms built into Kafka, you simply
specify a JAAS config with the combination of mechanisms you want and you
are done. If you are adding new mechanisms, it clearly can't be done
without some additional code to implement that mechanism. The callback and
login interfaces are configurable to ensure that (as a developer), you can
build custom mechanisms without modifying Kafka. SASL was designed
precisely to do this. But at the moment, Kafka restricts what type of
authentication you can perform with SASL.

Back to the questions:
1. *How does the proposed Login class interact with LoginContext and *
*LoginModule.*
Login class performs *LoginContext.login(). *This performs login for all
the login modules of that context (this is standard Java function). Login
is cached with a reference counter to keep it alive.

2. *Do we really need a pluggable Login class, when users **(administrators,
not developers) are more likely to want to specify LoginModules?*
Administrators specify *LoginModules. *Developers implement new mechanisms.
The pluggable *Login* class enables the developer to implement new
mechanisms without changing Kafka. If Kafka didn't have a default Kerberos
implementation and a user wanted to use Kerberos, pluggability of *Login* class
enables the user to include Kerberos without changing Kafka. Not that we
want another Kerberos, but Tao had a requirement to enable a custom
mechanism with functionality similar to Kerberos (see KIP-44). Pluggable
*Login* supports this use case without replacing the entire authentication
mechanism in Kafka.

*3. How will users get to specify multiple LoginModules?*
Standard Java JAAS configuration - see example above.


*4. It sounds like different LoginModules may need
different CallbackHandlers. Since we only allow a single
pluggableCallbackHandler class in the KIP-43 configuration, how will this
be handled?*
One LoginContext, one subject, one callback handler, multiple login
modules. This is the Java model for login. And it works well because
callback handlers tend to be fairly standard. Even if the smart card SASL
callbacks needed to do something wildly different and you wanted a broker
that combined this with Kerberos, you would implement a callback that
looked like the code below.

private CallbackHandler handler;

public void configure(Map configs, Mode mode, Subject
subject, String mechanism) {public void configure(Map configs,
Mode mode, Subject subject, String mechanism) {

switch (mechanism) {

case "GSSAPI" : handler = new GssapiCallbackHandler(); break
;

case "SMARTCARD" : handler = new SmartcardCallbackHandler();
break;

}

}


public void handle(Callback[] callbacks) throws IOException,
UnsupportedCallbackException {

handler.handle(callbacks);

}

But I don't see this as just an admin configuration. This is some code
developed alongside a new mechanism. Yes, you can add a new configuration
type in Kafka that maps mechanism to callback handler class to avoid the
switch statement. But that doesn't remove the need for new code altogether
since you still need to implement SmartcardCallbackHandler. Multiple
conflicting callback handlers within one broker to support complete
different behaviour is an edge case. Since the design supports
full flexibility in terms of functionality and you have to write new code
for new mechanisms anyway, this feels like the simplest solution.



On Tue, Mar 8, 2016 at 8:23 PM, Gwen Shapira  wrote:

> I guess what is unclear to me is:
>
> 1. How does the proposed Login class interact with LoginContext and
> LoginModule.
> 2. Do we really need a pluggable Login class, when users
> (administrators, not developers) are more likely to want to specify
> LoginModules?
> 3. How will users get to specify multiple LoginModules?
> 4. It sounds like different LoginModules may need different
> CallbackHandlers. Since we only allow a single pluggable
> CallbackHandler class in the KIP-43 

[jira] [Commented] (KAFKA-3312) Add a offsets methods to ZkUtils and replace relevant usages

[ASF GitHub Bot (JIRA)]

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

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

GitHub user vahidhashemian reopened a pull request:

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

KAFKA-3312: Add utility offset methods to ZkUtils



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

$ git pull https://github.com/vahidhashemian/kafka KAFKA-3312

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

https://github.com/apache/kafka/pull/1025.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 #1025


commit 8ee7711d11c6eb32f160c66202c54af2e17c3ed6
Author: Vahid Hashemian 
Date:   2016-03-05T00:47:05Z

KAFKA-3312: Add utility offset methods to ZkUtils




> Add a offsets methods to ZkUtils and replace relevant usages
> 
>
> Key: KAFKA-3312
> URL: https://issues.apache.org/jira/browse/KAFKA-3312
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Grant Henke
>Assignee: Vahid Hashemian
>
> There are many places in the code that manually build a zookeeper path and 
> get or update offsets. Moving this logic to a common location in ZkUtils 
> would be nice. 
> Ex:
> {code}
> zkUtils.readDataMaybeNull(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}")._1
> {code}
> {code}
>  zkUtils.readData(topicDirs.consumerOffsetDir + "/" + 
> topicAndPartition.partition)._1.toLong
> {code}
> {code}
> zkUtils.updatePersistentPath(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}",
>  partitionData.offset.toString)
> {code}



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

[GitHub] kafka pull request: KAFKA-3312: Add utility offset methods to ZkUt...

[vahidhashemian]

Github user vahidhashemian closed the pull request at:

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


---
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-3312: Add utility offset methods to ZkUt...

[vahidhashemian]

GitHub user vahidhashemian reopened a pull request:

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

KAFKA-3312: Add utility offset methods to ZkUtils



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

$ git pull https://github.com/vahidhashemian/kafka KAFKA-3312

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

https://github.com/apache/kafka/pull/1025.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 #1025


commit 8ee7711d11c6eb32f160c66202c54af2e17c3ed6
Author: Vahid Hashemian 
Date:   2016-03-05T00:47:05Z

KAFKA-3312: Add utility offset methods to ZkUtils




---
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-3312) Add a offsets methods to ZkUtils and replace relevant usages

[ASF GitHub Bot (JIRA)]

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

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

Github user vahidhashemian closed the pull request at:

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


> Add a offsets methods to ZkUtils and replace relevant usages
> 
>
> Key: KAFKA-3312
> URL: https://issues.apache.org/jira/browse/KAFKA-3312
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Grant Henke
>Assignee: Vahid Hashemian
>
> There are many places in the code that manually build a zookeeper path and 
> get or update offsets. Moving this logic to a common location in ZkUtils 
> would be nice. 
> Ex:
> {code}
> zkUtils.readDataMaybeNull(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}")._1
> {code}
> {code}
>  zkUtils.readData(topicDirs.consumerOffsetDir + "/" + 
> topicAndPartition.partition)._1.toLong
> {code}
> {code}
> zkUtils.updatePersistentPath(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}",
>  partitionData.offset.toString)
> {code}



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

[jira] [Updated] (KAFKA-3352) Avoid DNS reverse lookups

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3352:
---
Status: Patch Available  (was: Open)

> Avoid DNS reverse lookups
> -
>
> Key: KAFKA-3352
> URL: https://issues.apache.org/jira/browse/KAFKA-3352
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ismael Juma
>Assignee: Ismael Juma
> Fix For: 0.10.0.0
>
>
> We use `InetSocketAddress.getHostName` in `Cluster.bootstrap` and 
> `SocketServer` This call can cause a reverse DNS look-up if an IP address is 
> used in the configuration. `getHostString` avoids the reverse DNS look-up.
> `InetAddress.getHostName` has the same issue and it's used in 
> `SaslChannelBuilder`, `SslChannelBuilder` and `ZookeeperConsumerConnector`. I 
> am not changing these because the the former two actually require the FQDN 
> while the latter is on its way to deprecation.



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

[jira] [Updated] (KAFKA-3352) Avoid DNS reverse lookups

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3352:
---
Fix Version/s: 0.10.0.0

> Avoid DNS reverse lookups
> -
>
> Key: KAFKA-3352
> URL: https://issues.apache.org/jira/browse/KAFKA-3352
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ismael Juma
>Assignee: Ismael Juma
> Fix For: 0.10.0.0
>
>
> We use `InetSocketAddress.getHostName` in `Cluster.bootstrap` and 
> `SocketServer` This call can cause a reverse DNS look-up if an IP address is 
> used in the configuration. `getHostString` avoids the reverse DNS look-up.
> `InetAddress.getHostName` has the same issue and it's used in 
> `SaslChannelBuilder`, `SslChannelBuilder` and `ZookeeperConsumerConnector`. I 
> am not changing these because the the former two actually require the FQDN 
> while the latter is on its way to deprecation.



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

[jira] [Commented] (KAFKA-3352) Avoid DNS reverse lookups

[ASF GitHub Bot (JIRA)]

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

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

GitHub user ijuma opened a pull request:

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

KAFKA-3352: Avoid DNS reverse lookups

By using `getHostString` (introduced in Java 7) instead of `getHostName`.

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

$ git pull https://github.com/ijuma/kafka 
kafka-3352-avoid-dns-reverse-look-ups

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

https://github.com/apache/kafka/pull/1030.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 #1030


commit 3da9c7ec7e137efab2ba300deb0a94918188eb5b
Author: Ismael Juma 
Date:   2016-03-08T22:11:36Z

Use `getHostString` instead of `getHostName` to avoid reverse DNS look-ups




> Avoid DNS reverse lookups
> -
>
> Key: KAFKA-3352
> URL: https://issues.apache.org/jira/browse/KAFKA-3352
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ismael Juma
>Assignee: Ismael Juma
>
> We use `InetSocketAddress.getHostName` in `Cluster.bootstrap` and 
> `SocketServer` This call can cause a reverse DNS look-up if an IP address is 
> used in the configuration. `getHostString` avoids the reverse DNS look-up.
> `InetAddress.getHostName` has the same issue and it's used in 
> `SaslChannelBuilder`, `SslChannelBuilder` and `ZookeeperConsumerConnector`. I 
> am not changing these because the the former two actually require the FQDN 
> while the latter is on its way to deprecation.



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

[GitHub] kafka pull request: KAFKA-3352: Avoid DNS reverse lookups

[ijuma]

GitHub user ijuma opened a pull request:

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

KAFKA-3352: Avoid DNS reverse lookups

By using `getHostString` (introduced in Java 7) instead of `getHostName`.

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

$ git pull https://github.com/ijuma/kafka 
kafka-3352-avoid-dns-reverse-look-ups

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

https://github.com/apache/kafka/pull/1030.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 #1030


commit 3da9c7ec7e137efab2ba300deb0a94918188eb5b
Author: Ismael Juma 
Date:   2016-03-08T22:11:36Z

Use `getHostString` instead of `getHostName` to avoid reverse DNS look-ups




---
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] [Updated] (KAFKA-3197) Producer can send message out of order even when in flight request is set to 1.

[Joel Koshy (JIRA)]

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

Joel Koshy updated KAFKA-3197:
--
Resolution: Fixed
Status: Resolved  (was: Patch Available)

Pushed to trunk.

> Producer can send message out of order even when in flight request is set to 
> 1.
> ---
>
> Key: KAFKA-3197
> URL: https://issues.apache.org/jira/browse/KAFKA-3197
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, producer 
>Affects Versions: 0.9.0.0
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> The issue we saw is following:
> 1. Producer send message 0 to topic-partition-0 on broker A. The in-flight 
> request to broker A is 1.
> 2. The request is somehow lost
> 3. Producer refreshed its topic metadata and found leader of 
> topic-partition-0 migrated from broker A to broker B.
> 4. Because there is no in-flight request to broker B. All the subsequent 
> messages to topic-partition-0 in the record accumulator are sent to broker B.
> 5. Later on when the request in step (1) times out, message 0 will be retried 
> and sent to broker B. At this point, all the later messages has already been 
> sent, so we have re-order.



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

[jira] [Updated] (KAFKA-3352) Avoid DNS reverse lookups

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3352:
---
Description: 
We use `InetSocketAddress.getHostName` in `Cluster.bootstrap` and 
`SocketServer` This call can cause a reverse DNS look-up if an IP address is 
used in the configuration. `getHostString` avoids the reverse DNS look-up.

`InetAddress.getHostName` has the same issue and it's used in 
`SaslChannelBuilder`, `SslChannelBuilder` and `ZookeeperConsumerConnector`. I 
am not changing these because the the former two actually require the FQDN 
while the latter is on its way to deprecation.

  was:We use `InetSocketAddress.getHostName` in Cluster.bootstrap. This call 
can cause a reverse DNS look-up if an IP address is used in the configuration. 
`getHostString` avoids the reverse DNS look-up.


> Avoid DNS reverse lookups
> -
>
> Key: KAFKA-3352
> URL: https://issues.apache.org/jira/browse/KAFKA-3352
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ismael Juma
>Assignee: Ismael Juma
>
> We use `InetSocketAddress.getHostName` in `Cluster.bootstrap` and 
> `SocketServer` This call can cause a reverse DNS look-up if an IP address is 
> used in the configuration. `getHostString` avoids the reverse DNS look-up.
> `InetAddress.getHostName` has the same issue and it's used in 
> `SaslChannelBuilder`, `SslChannelBuilder` and `ZookeeperConsumerConnector`. I 
> am not changing these because the the former two actually require the FQDN 
> while the latter is on its way to deprecation.



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

[GitHub] kafka pull request: KAFKA-3197 Fix producer sending records out of...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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-3197) Producer can send message out of order even when in flight request is set to 1.

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Producer can send message out of order even when in flight request is set to 
> 1.
> ---
>
> Key: KAFKA-3197
> URL: https://issues.apache.org/jira/browse/KAFKA-3197
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, producer 
>Affects Versions: 0.9.0.0
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> The issue we saw is following:
> 1. Producer send message 0 to topic-partition-0 on broker A. The in-flight 
> request to broker A is 1.
> 2. The request is somehow lost
> 3. Producer refreshed its topic metadata and found leader of 
> topic-partition-0 migrated from broker A to broker B.
> 4. Because there is no in-flight request to broker B. All the subsequent 
> messages to topic-partition-0 in the record accumulator are sent to broker B.
> 5. Later on when the request in step (1) times out, message 0 will be retried 
> and sent to broker B. At this point, all the later messages has already been 
> sent, so we have re-order.



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

[jira] [Updated] (KAFKA-3352) Avoid DNS reverse lookups

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3352:
---
Summary: Avoid DNS reverse lookups  (was: Avoid DNS reverse lookup in 
Cluster.bootstrap)

> Avoid DNS reverse lookups
> -
>
> Key: KAFKA-3352
> URL: https://issues.apache.org/jira/browse/KAFKA-3352
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ismael Juma
>Assignee: Ismael Juma
>
> We use `InetSocketAddress.getHostName` in Cluster.bootstrap. This call can 
> cause a reverse DNS look-up if an IP address is used in the configuration. 
> `getHostString` avoids the reverse DNS look-up.



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

[jira] [Created] (KAFKA-3352) Avoid DNS reverse lookup in Cluster.bootstrap

[Ismael Juma (JIRA)]

Ismael Juma created KAFKA-3352:
--

 Summary: Avoid DNS reverse lookup in Cluster.bootstrap
 Key: KAFKA-3352
 URL: https://issues.apache.org/jira/browse/KAFKA-3352
 Project: Kafka
  Issue Type: Improvement
Reporter: Ismael Juma
Assignee: Ismael Juma


We use `InetSocketAddress.getHostName` in Cluster.bootstrap. This call can 
cause a reverse DNS look-up if an IP address is used in the configuration. 
`getHostString` avoids the reverse DNS look-up.



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

[jira] [Commented] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff commented on KAFKA-3323:
---

[~gwenshap] [~ijuma] https://github.com/tubemogul/kafka-logsplitter
Thank you guys!

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Commented] (KAFKA-3342) https://cwiki.apache.org/confluence/display/KAFKA/Log+Compaction has log.cleaner.min.cleanable.ratio listed twice in error

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian commented on KAFKA-3342:


I believe the second one should read _log.cleaner.threads_.

> https://cwiki.apache.org/confluence/display/KAFKA/Log+Compaction has 
> log.cleaner.min.cleanable.ratio listed twice in error
> --
>
> Key: KAFKA-3342
> URL: https://issues.apache.org/jira/browse/KAFKA-3342
> Project: Kafka
>  Issue Type: Bug
>Reporter: Michael Graff
>Priority: Minor
>
> https://cwiki.apache.org/confluence/display/KAFKA/Log+Compaction has 
> log.cleaner.min.cleanable.ratio listed both for what it should describe, and 
> duplicated again for the setting for the number of threads.  This is in the 
> table near the end of the page.



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

[jira] [Commented] (KAFKA-2066) Replace FetchRequest / FetchResponse with their org.apache.kafka.common.requests equivalents

[Grant Henke (JIRA)]

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

Grant Henke commented on KAFKA-2066:


[~dajac] Are you still working on this?

> Replace FetchRequest / FetchResponse with their 
> org.apache.kafka.common.requests equivalents
> 
>
> Key: KAFKA-2066
> URL: https://issues.apache.org/jira/browse/KAFKA-2066
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Gwen Shapira
>Assignee: David Jacot
>
> Replace FetchRequest / FetchResponse with their 
> org.apache.kafka.common.requests equivalents.
> Note that they can't be completely removed until we deprecate the 
> SimpleConsumer API (and it will require very careful patchwork for the places 
> where core modules actually use the SimpleConsumer API).
> This also requires a solution on how to stream from memory-mapped files 
> (similar to what existing code does with FileMessageSet. 



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

[jira] [Commented] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Ismael Juma (JIRA)]

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

Ismael Juma commented on KAFKA-3323:


Yes, thank you Michael. :)

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Commented] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Gwen Shapira (JIRA)]

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

Gwen Shapira commented on KAFKA-3323:
-

You rock! Thank you.

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Updated] (KAFKA-3306) Change metdata response to include required additional fields

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3306:
---
Description: 
This includes:
- Rack information from KAFKA-1215 (KIP-36)
- A boolean to indicate if a topic is marked for deletion
- A boolean to indicate if a topic is an internal topic
- An indicator of the current controller broker

  was:
This includes:
- Rack information from KAFKA-1215 (KIP-36)
- A boolean to indicate if a topic is marked for deletion
- A boolean to indicate if a topic is an internal topic


> Change metdata response to include required additional fields
> -
>
> Key: KAFKA-3306
> URL: https://issues.apache.org/jira/browse/KAFKA-3306
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
>Assignee: Grant Henke
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> This includes:
> - Rack information from KAFKA-1215 (KIP-36)
> - A boolean to indicate if a topic is marked for deletion
> - A boolean to indicate if a topic is an internal topic
> - An indicator of the current controller broker



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

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Gwen Shapira]

I guess what is unclear to me is:

1. How does the proposed Login class interact with LoginContext and
LoginModule.
2. Do we really need a pluggable Login class, when users
(administrators, not developers) are more likely to want to specify
LoginModules?
3. How will users get to specify multiple LoginModules?
4. It sounds like different LoginModules may need different
CallbackHandlers. Since we only allow a single pluggable
CallbackHandler class in the KIP-43 configuration, how will this be
handled?

In other words, suppose I want my Kafka installation to support both
Kerberos and SmartCard mechanisms, can you tell me how to configure
Kafka? (as admin, not developer)?

Gwen



On Tue, Mar 8, 2016 at 11:46 AM, Rajini Sivaram
 wrote:
> Gwen,
>
> I am not sure I understand your concern. Java login contexts support
> multiple login modules precisely to support this type of real-world
> scenario.
>
> From Javadoc of *javax.security.auth.login.LoginContext*:
>
> *In addition to supporting pluggable authentication, this class also
> supports the notion of stacked authentication. Applications may be
> configured to use more than one LoginModule. For example, one could
> configure both a Kerberos LoginModule and a smart card LoginModule under an
> application. *
>
> * A typical caller instantiates a LoginContext with a name and a
> CallbackHandler. LoginContext uses the name as the index into a
> Configuration to determine which LoginModules should be used, and which
> ones must succeed in order for the overall authentication to succeed. The
> CallbackHandler is passed to the underlying LoginModules so they may
> communicate and interact with users (prompting for a username and password
> via a graphical user interface, for example).*
>
> If it is good enough for Java, surely it must be good enough for Kafka. We
> don't expect users to directly use existing classes when they want to
> support complex new mechanisms. Login is not a standard interface, and we
> want to make callback handlers configurable. So we do require users to wrap
> their existing classes in our interface. And the interfaces make it
> possible to plugin any number of mechanisms. Am I missing something?
>
>
>
> On Tue, Mar 8, 2016 at 4:49 PM, Gwen Shapira  wrote:
>
>> Yes, I understand that.
>>
>> However, because the current configuration only allows a single
>> callback and a single login module, I need to supply a class that
>> supports callbacks or logins for every single mechanism that I need to
>> support. I question whether this is applicable in real-world scenario
>> where a user may need to support mechanisms that were written by
>> different vendors and where a single class that implements all the
>> necessary modules or callsbacks is very unlikely to exist.
>>
>> I agree that supporting a list of classes and dynamically using the
>> correct callbacks and modules based on configuration is complex, but I
>> don't think this feature is useful enough without it.
>>
>> Gwen
>>
>> On Tue, Mar 8, 2016 at 2:40 AM, Rajini Sivaram
>>  wrote:
>> > Gwen,
>> >
>> > Sorry if this is not clear in the KIP. The goal of the KIP is to enable
>> new
>> > mechanisms to be added without any changes to Apache Kafka. The
>> > configuration proposed by the KIP is based on the configurability of the
>> > Java SASL API and what it would take to add a new mechanism similar to
>> > Kerberos/PLAIN/Digest-MD5 and the requirements that Tao raised for custom
>> > authentication.
>> >
>> > The PR includes a test for Digest-MD5 which is not included in Kafka, but
>> > is tested using the configuration options that allow new mechanisms to be
>> > added.
>> >
>> > If you want to add a more complex mechanism which requires integration
>> with
>> > some external authentication libraries for login or callback handlers,
>> you
>> > can do that by configuring your own callback handler and login
>> > implementation. The login class allows you to add additional threads (for
>> > instance for token refresh), while the callback handler extends the
>> > standard Java CallbackHandler interface, enabling you to override any of
>> > the default callbacks in Kafka to provide your own behaviour. If you want
>> > to add your own login modules, SASL server or SASL client implementation
>> > for a mechanism that is not supported in your JDK, you can do that by
>> > adding new Java security modules, using the built-in pluggability of Java
>> > security.
>> >
>> > And finally, if you want to enable multiple mechanisms in your Kafka
>> > brokers, you can specify a callback handler class that either provides
>> the
>> > handlers for all the mechanisms (as the PR does for Kerberos and PLAIN)
>> or
>> > a delegator that calls different callback handlers based on the
>> mechanism.
>> > And for login, you provide a single or a compound class that provides any
>> > additional login logic for all the login modules that 

[jira] [Commented] (KAFKA-3312) Add a offsets methods to ZkUtils and replace relevant usages

[ASF GitHub Bot (JIRA)]

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

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

GitHub user vahidhashemian reopened a pull request:

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

KAFKA-3312: Add utility offset methods to ZkUtils



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

$ git pull https://github.com/vahidhashemian/kafka KAFKA-3312

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

https://github.com/apache/kafka/pull/1025.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 #1025


commit 8ee7711d11c6eb32f160c66202c54af2e17c3ed6
Author: Vahid Hashemian 
Date:   2016-03-05T00:47:05Z

KAFKA-3312: Add utility offset methods to ZkUtils




> Add a offsets methods to ZkUtils and replace relevant usages
> 
>
> Key: KAFKA-3312
> URL: https://issues.apache.org/jira/browse/KAFKA-3312
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Grant Henke
>Assignee: Vahid Hashemian
>
> There are many places in the code that manually build a zookeeper path and 
> get or update offsets. Moving this logic to a common location in ZkUtils 
> would be nice. 
> Ex:
> {code}
> zkUtils.readDataMaybeNull(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}")._1
> {code}
> {code}
>  zkUtils.readData(topicDirs.consumerOffsetDir + "/" + 
> topicAndPartition.partition)._1.toLong
> {code}
> {code}
> zkUtils.updatePersistentPath(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}",
>  partitionData.offset.toString)
> {code}



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

[GitHub] kafka pull request: KAFKA-3312: Add utility offset methods to ZkUt...

[vahidhashemian]

Github user vahidhashemian closed the pull request at:

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


---
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-3312: Add utility offset methods to ZkUt...

[vahidhashemian]

GitHub user vahidhashemian reopened a pull request:

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

KAFKA-3312: Add utility offset methods to ZkUtils



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

$ git pull https://github.com/vahidhashemian/kafka KAFKA-3312

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

https://github.com/apache/kafka/pull/1025.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 #1025


commit 8ee7711d11c6eb32f160c66202c54af2e17c3ed6
Author: Vahid Hashemian 
Date:   2016-03-05T00:47:05Z

KAFKA-3312: Add utility offset methods to ZkUtils




---
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] [Updated] (KAFKA-3306) Change metdata response to include required additional fields

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3306:
---
Priority: Critical  (was: Major)

> Change metdata response to include required additional fields
> -
>
> Key: KAFKA-3306
> URL: https://issues.apache.org/jira/browse/KAFKA-3306
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
>Assignee: Grant Henke
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> This includes:
> - Rack information from KAFKA-1215 (KIP-36)
> - A boolean to indicate if a topic is marked for deletion
> - A boolean to indicate if a topic is an internal topic



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

[jira] [Updated] (KAFKA-3306) Change metdata response to include required additional fields

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3306:
---
Fix Version/s: 0.10.0.0

> Change metdata response to include required additional fields
> -
>
> Key: KAFKA-3306
> URL: https://issues.apache.org/jira/browse/KAFKA-3306
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
>Assignee: Grant Henke
> Fix For: 0.10.0.0
>
>
> This includes:
> - Rack information from KAFKA-1215 (KIP-36)
> - A boolean to indicate if a topic is marked for deletion
> - A boolean to indicate if a topic is an internal topic



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

[jira] [Resolved] (KAFKA-3269) Update topic metadata request to included needed information for polling

[Grant Henke (JIRA)]

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

Grant Henke resolved KAFKA-3269.

Resolution: Duplicate

> Update topic metadata request to included needed information for polling
> 
>
> Key: KAFKA-3269
> URL: https://issues.apache.org/jira/browse/KAFKA-3269
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
>Assignee: Grant Henke
>
> Add markedForDelete flag to topic metadata and any other information required 
> to poll for the status of admin requests in KIP-4. This jira will be updated 
> as the server implementations progress. 
> This change may want to be done at a similar time to adding the rack 
> information for KAFKA-1215.



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

[jira] [Updated] (KAFKA-3306) Change metdata response to include required additional fields

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3306:
---
Description: 
This includes:
- Rack information from KAFKA-1215 (KIP-36)
- A boolean to indicate if a topic is marked for deletion
- A boolean to indicate if a topic is an internal topic

  was:
This includes:
- Rack information from KAFKA-1215 (KIP-36)
- A boolean to indicate if a topic is marked for deletion


> Change metdata response to include required additional fields
> -
>
> Key: KAFKA-3306
> URL: https://issues.apache.org/jira/browse/KAFKA-3306
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
>Assignee: Grant Henke
>
> This includes:
> - Rack information from KAFKA-1215 (KIP-36)
> - A boolean to indicate if a topic is marked for deletion
> - A boolean to indicate if a topic is an internal topic



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

[jira] [Updated] (KAFKA-3266) List/Alter Acls - protocol and server side implementation

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3266:
---
Status: Patch Available  (was: In Progress)

> List/Alter Acls - protocol and server side implementation
> -
>
> Key: KAFKA-3266
> URL: https://issues.apache.org/jira/browse/KAFKA-3266
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
>Assignee: Grant Henke
> Fix For: 0.10.0.0
>
>




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

[jira] [Updated] (KAFKA-3266) List/Alter Acls - protocol and server side implementation

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3266:
---
Fix Version/s: 0.10.0.0

> List/Alter Acls - protocol and server side implementation
> -
>
> Key: KAFKA-3266
> URL: https://issues.apache.org/jira/browse/KAFKA-3266
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
>Assignee: Grant Henke
> Fix For: 0.10.0.0
>
>




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

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Rajini Sivaram]

Gwen,

I am not sure I understand your concern. Java login contexts support
multiple login modules precisely to support this type of real-world
scenario.

>From Javadoc of *javax.security.auth.login.LoginContext*:

*In addition to supporting pluggable authentication, this class also
supports the notion of stacked authentication. Applications may be
configured to use more than one LoginModule. For example, one could
configure both a Kerberos LoginModule and a smart card LoginModule under an
application. *

* A typical caller instantiates a LoginContext with a name and a
CallbackHandler. LoginContext uses the name as the index into a
Configuration to determine which LoginModules should be used, and which
ones must succeed in order for the overall authentication to succeed. The
CallbackHandler is passed to the underlying LoginModules so they may
communicate and interact with users (prompting for a username and password
via a graphical user interface, for example).*

If it is good enough for Java, surely it must be good enough for Kafka. We
don't expect users to directly use existing classes when they want to
support complex new mechanisms. Login is not a standard interface, and we
want to make callback handlers configurable. So we do require users to wrap
their existing classes in our interface. And the interfaces make it
possible to plugin any number of mechanisms. Am I missing something?



On Tue, Mar 8, 2016 at 4:49 PM, Gwen Shapira  wrote:

> Yes, I understand that.
>
> However, because the current configuration only allows a single
> callback and a single login module, I need to supply a class that
> supports callbacks or logins for every single mechanism that I need to
> support. I question whether this is applicable in real-world scenario
> where a user may need to support mechanisms that were written by
> different vendors and where a single class that implements all the
> necessary modules or callsbacks is very unlikely to exist.
>
> I agree that supporting a list of classes and dynamically using the
> correct callbacks and modules based on configuration is complex, but I
> don't think this feature is useful enough without it.
>
> Gwen
>
> On Tue, Mar 8, 2016 at 2:40 AM, Rajini Sivaram
>  wrote:
> > Gwen,
> >
> > Sorry if this is not clear in the KIP. The goal of the KIP is to enable
> new
> > mechanisms to be added without any changes to Apache Kafka. The
> > configuration proposed by the KIP is based on the configurability of the
> > Java SASL API and what it would take to add a new mechanism similar to
> > Kerberos/PLAIN/Digest-MD5 and the requirements that Tao raised for custom
> > authentication.
> >
> > The PR includes a test for Digest-MD5 which is not included in Kafka, but
> > is tested using the configuration options that allow new mechanisms to be
> > added.
> >
> > If you want to add a more complex mechanism which requires integration
> with
> > some external authentication libraries for login or callback handlers,
> you
> > can do that by configuring your own callback handler and login
> > implementation. The login class allows you to add additional threads (for
> > instance for token refresh), while the callback handler extends the
> > standard Java CallbackHandler interface, enabling you to override any of
> > the default callbacks in Kafka to provide your own behaviour. If you want
> > to add your own login modules, SASL server or SASL client implementation
> > for a mechanism that is not supported in your JDK, you can do that by
> > adding new Java security modules, using the built-in pluggability of Java
> > security.
> >
> > And finally, if you want to enable multiple mechanisms in your Kafka
> > brokers, you can specify a callback handler class that either provides
> the
> > handlers for all the mechanisms (as the PR does for Kerberos and PLAIN)
> or
> > a delegator that calls different callback handlers based on the
> mechanism.
> > And for login, you provide a single or a compound class that provides any
> > additional login logic for all the login modules that you have specified
> > for KafkaServer.
> >
> > Hope that helps.
> >
> >
> > On Tue, Mar 8, 2016 at 1:17 AM, Gwen Shapira  wrote:
> >
> >> Can you explain the process for a adding a new mechanism based on
> current
> >> KIP?
> >>
> >> My thought is that if it requires modifying Apache Kafka code, it is
> >> not pluggable enough.
> >>
> >> On Mon, Mar 7, 2016 at 4:04 PM, Rajini Sivaram
> >>  wrote:
> >> > Gwen,
> >> >
> >> > In cases where you want completely different callbacks for different
> >> > mechanisms, I was thinking that the choice would be between a map of
> >> > classes (mechanism -> callbackHandler class) or a delegator class that
> >> > chooses the appropriate callback handler class based on mechanism. I
> >> chose
> >> > the latter since it makes it easier to configure in Kafka. Since we
> >> create
> >> > a 

[jira] [Commented] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Gwen Shapira (JIRA)]

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

Gwen Shapira commented on KAFKA-3323:
-

[~michael.schiff] Yeah, i was thinking that you can host this as a stand alone 
tool in Github, and link to it here (and in KAFKA-2024).
This way people with older versions will be able to find and use your solution, 
but we won't package something that is no longer useful with new releases.

Makes sense?

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

Re: [VOTE] KIP-33 - Add a time based log index to Kafka

[Becket Qin]

I updated the wiki to make the following change: Instead of maintaining a
globally monotonically increasing time index. We only make sure the time
index for each log segment is monotonically increasing.

By doing that everything seems much simpler. we avoid empty time index.
Enforcing log retention becomes easier. And searching by timestamp is not
complicated.

Thanks,

Jiangjie (Becket) Qin

On Mon, Mar 7, 2016 at 2:52 PM, Becket Qin  wrote:

> Hi Jun,
>
> What do you think about the above solution? I am trying to include KIP-33
> into 0.10.0 because the log retention has been a long pending issue.
>
> Thanks,
>
> Jiangjie (Becket) Qin
>
> On Tue, Mar 1, 2016 at 8:18 PM, Becket Qin  wrote:
>
>> Hi Jun,
>>
>> I see. If we only use index.interval.bytes, the time index entry will be
>> inserted when (1) the largest timestamp is in this segment AND (2) at least
>> index.interval.bytes have been appended since last time index entry
>> insertion.
>> In this case (1) becomes implicit instead of having an explicit threshold
>> of time.index.interval.ms. This should work fine.
>>
>> For 1, the current proposal is actually intended to use the offsets of
>> the messages instead of the file position. The reason is that
>> OffsetRequest(ListOffsetRequest) gives back a list of offsets. Having
>> offsets instead of file position is more convenient in that case. So far we
>> don't have an interface for consumer to directly consume from a given
>> timestamp. Supposedly we will have to first return the offset to the
>> consumer then the consumer can issue a fetch request. But this does mean
>> that we need to look up the index twice if we want to search to a
>> particular message using timestamp.
>>
>> For 2, that is a good point. It looks we have to persist the max
>> timestamp of each segment. I am thinking of reserving the first time index
>> entry in each time index file. When broker shuts down or rolls out a new
>> segment, we persist the max timestamp in the segment by writing a time
>> index entry to the first entry. During timestamp search we will ignore the
>> first time index entry. So the log retention will always be able to know
>> for sure if a log segment is supposed to be deleted or not by looking at
>> the first entry of time index.
>>
>> Thanks,
>>
>> Jiangjie (Becket) Qin
>>
>>
>>
>>
>> On Tue, Mar 1, 2016 at 4:30 PM, Jun Rao  wrote:
>>
>>> Hi, Jiangjie,
>>>
>>> I was thinking perhaps just reusing index.interval.bytes is enough. Not
>>> sure if there is much value in adding an additional
>>> time.index.interval.ms.
>>>
>>> For 1, the timestamp index has entries of timestamp -> file position. So,
>>> there is actually no offset in the index, right?
>>>
>>> For 2, what you said makes sense for time-based retention. Does that
>>> apply
>>> if the retention is trigged by size? The difference here is that we can't
>>> assume all segments with messages of timestamp smaller than the latest
>>> timestamp will be deleted after the message with the latest timestamp is
>>> deleted.
>>>
>>> Thanks,
>>>
>>> Jun
>>>
>>> On Tue, Mar 1, 2016 at 1:00 PM, Becket Qin  wrote:
>>>
>>> > Hi Jun,
>>> >
>>> > Rolling out a new segment when the time index is full sounds good. So
>>> both
>>> > time index and offset index will be sharing the configuration of max
>>> index
>>> > size.
>>> > If we do that, do you think we still want to reuse
>>> index.interval.bytes? If
>>> > we don't, the risk is that in some corner cases, we might end up with
>>> many
>>> > small segments. (e.g. small time.index.interval.ms with small max
>>> index
>>> > size). But this is probably more of a misconfiguration.
>>> >
>>> > 2. If the broker is still running when all the segments except the
>>> active
>>> > segment is deleted, we will have an in memory latest timestamp. So
>>> that is
>>> > not a problem.
>>> >
>>> > In another case, if a broker boots up and sees only one segment with an
>>> > empty time index file, we can scan the active segment and rebuild the
>>> time
>>> > index.  i.e. we do not need to care about the previous largest
>>> timestamp
>>> > but simply start over. (We need to scan the active segment because it
>>> is
>>> > possible that the last message appended to the log has a timestamp not
>>> > expired, but the broker died before inserting the time index entry for
>>> > it.). If all the messages in the active segment has expired, we should
>>> roll
>>> > out a new segment and reset the latest timetamp to -1.
>>> > The principal here is that we will try to build the time indices for
>>> the
>>> > existing segments that have not expired. If the message with previously
>>> > latest timestamp has already been deleted, there is no need to remember
>>> > that any more.
>>> >
>>> > That said, I believe this corner case is really because user is not
>>> > configuring the acceptable time difference threshold appropriately.
>>> >
>>> > Thanks,
>>> >
>>> > 

[jira] [Comment Edited] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff edited comment on KAFKA-3323 at 3/8/16 7:33 PM:
---

[~ijuma] I reopened to submit the patch, but this didn't seem like the correct 
action, as this utility is currently just a standalone program, not a part of 
the Kafka distribution. Should I convert this utility into something that can 
be run like the other utilities (e.g. dump log segments)?  Once everyone has 
upgraded, this tool will not be useful anymore, so I don't know that it belongs 
in the distribution.


was (Author: michael.schiff):
[~ijuma] I went to submit the patch, but this didn't seem like the correct 
action, as this utility is currently just a standalone program, not a part of 
the Kafka distribution. Should I convert this utility into something that can 
be run like the other utilities (e.g. dump log segments)?  Once everyone has 
upgraded, this tool will not be useful anymore, so I don't know that it belongs 
in the distribution.

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Commented] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff commented on KAFKA-3323:
---

[~ijuma] I went to submit the patch, but this didn't seem like the correct 
action, as this utility is currently just a standalone program, not a part of 
the Kafka distribution. Should I convert this utility into something that can 
be run like the other utilities (e.g. dump log segments)?  Once everyone has 
upgraded, this tool will not be useful anymore, so I don't know that it belongs 
in the distribution.

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Comment Edited] (KAFKA-1843) Metadata fetch/refresh in new producer should handle all node connection states gracefully

[venu (JIRA)]

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

venu edited comment on KAFKA-1843 at 3/8/16 7:11 PM:
-

I am having similar problem , when partition is down, metadata is still 
returning the downed partitions as up in available partitions List. how to 
refresh/update the metadata when partition is down.  (I am using Kafka 0.8.2.1V)


was (Author: venu516):
I am having similar problem , when partition is down, metadata is still 
returning the downed partitions as up. how to refresh/update the metadata when 
partition is down.  (I am using Kafka 0.8.2.1V)

> Metadata fetch/refresh in new producer should handle all node connection 
> states gracefully
> --
>
> Key: KAFKA-1843
> URL: https://issues.apache.org/jira/browse/KAFKA-1843
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, producer 
>Affects Versions: 0.8.2.0
>Reporter: Ewen Cheslack-Postava
>Assignee: Eno Thereska
>Priority: Blocker
>  Labels: patch
> Fix For: 0.8.2.1
>
>
> KAFKA-1642 resolved some issues with the handling of broker connection states 
> to avoid high CPU usage, but made the minimal fix rather than the ideal one. 
> The code for handling the metadata fetch is difficult to get right because it 
> has to handle a lot of possible connectivity states and failure modes across 
> all the known nodes. It also needs to correctly integrate with the 
> surrounding event loop, providing correct poll() timeouts to both avoid busy 
> looping and make sure it wakes up and tries new nodes in the face of both 
> connection and request failures.
> A patch here should address a few issues:
> 1. Make sure connection timeouts, as implemented in KAFKA-1842, are cleanly 
> integrated. This mostly means that when a connecting node is selected to 
> fetch metadata from, that the code notices that and sets the next timeout 
> based on the connection timeout rather than some other backoff.
> 2. Rethink the logic and naming of NetworkClient.leastLoadedNode. That method 
> actually takes into account a) the current connectivity of each node, b) 
> whether the node had a recent connection failure, c) the "load" in terms of 
> in flight requests. It also needs to ensure that different clients don't use 
> the same ordering across multiple calls (which is already addressed in the 
> current code by nodeIndexOffset) and that we always eventually try all nodes 
> in the face of connection failures (which isn't currently handled by 
> leastLoadedNode and probably cannot be without tracking additional state). 
> This method also has to work for new consumer use cases even though it is 
> currently only used by the new producer's metadata fetch. Finally it has to 
> properly handle when other code calls initiateConnect() since the normal path 
> for sending messages also initiates connections.
> We can already say that there is an order of preference given a single call 
> (as follows), but making this work across multiple calls when some initial 
> choices fail to connect or return metadata *and* connection states may be 
> changing is much more difficult.
>  * Connected, zero in flight requests - the request can be sent immediately
>  * Connecting node - it will hopefully be connected very soon and by 
> definition has no in flight requests
>  * Disconnected - same reasoning as for a connecting node
>  * Connected, > 0 in flight requests - we consider any # of in flight 
> requests as a big enough backlog to delay the request a lot.
> We could use an approach that better accounts for # of in flight requests 
> rather than just turning it into a boolean variable, but that probably 
> introduces much more complexity than it is worth.
> 3. The most difficult case to handle so far has been when leastLoadedNode 
> returns a disconnected node to maybeUpdateMetadata as its best option. 
> Properly handling the two resulting cases (initiateConnect fails immediately 
> vs. taking some time to possibly establish the connection) is tricky.
> 4. Consider optimizing for the failure cases. The most common cases are when 
> you already have an active connection and can immediately get the metadata or 
> you need to establish a connection, but the connection and metadata 
> request/response happen very quickly. These common cases are infrequent 
> enough (default every 5 min) that establishing an extra connection isn't a 
> big deal as long as it's eventually cleaned up. The edge cases, like network 
> partitions where some subset of nodes become unreachable for a long period, 
> are harder to reason about but we should be sure we will always be able to 
> gracefully recover from them.
> KAFKA-1642 

[jira] [Updated] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff updated KAFKA-3323:
--
Status: Open  (was: Patch Available)

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.2.1, 0.8.1.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Issue Comment Deleted] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff updated KAFKA-3323:
--
Comment: was deleted

(was: Re-opening so that I can attach the log-splitter patch)

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Comment Edited] (KAFKA-1843) Metadata fetch/refresh in new producer should handle all node connection states gracefully

[venu (JIRA)]

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

venu edited comment on KAFKA-1843 at 3/8/16 7:08 PM:
-

I am having similar problem , when partition is down, metadata is still 
returning the downed partitions as up. how to refresh/update the metadata when 
partition is down.  (I am using Kafka 0.8.2.1V)


was (Author: venu516):
I am having similar problem , when partition is down, metadata is still 
returning the downed partitions as up. how to refresh/update the metadata when 
partition is down. Using Kafka 0.8.2

> Metadata fetch/refresh in new producer should handle all node connection 
> states gracefully
> --
>
> Key: KAFKA-1843
> URL: https://issues.apache.org/jira/browse/KAFKA-1843
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, producer 
>Affects Versions: 0.8.2.0
>Reporter: Ewen Cheslack-Postava
>Assignee: Eno Thereska
>Priority: Blocker
>  Labels: patch
> Fix For: 0.8.2.1
>
>
> KAFKA-1642 resolved some issues with the handling of broker connection states 
> to avoid high CPU usage, but made the minimal fix rather than the ideal one. 
> The code for handling the metadata fetch is difficult to get right because it 
> has to handle a lot of possible connectivity states and failure modes across 
> all the known nodes. It also needs to correctly integrate with the 
> surrounding event loop, providing correct poll() timeouts to both avoid busy 
> looping and make sure it wakes up and tries new nodes in the face of both 
> connection and request failures.
> A patch here should address a few issues:
> 1. Make sure connection timeouts, as implemented in KAFKA-1842, are cleanly 
> integrated. This mostly means that when a connecting node is selected to 
> fetch metadata from, that the code notices that and sets the next timeout 
> based on the connection timeout rather than some other backoff.
> 2. Rethink the logic and naming of NetworkClient.leastLoadedNode. That method 
> actually takes into account a) the current connectivity of each node, b) 
> whether the node had a recent connection failure, c) the "load" in terms of 
> in flight requests. It also needs to ensure that different clients don't use 
> the same ordering across multiple calls (which is already addressed in the 
> current code by nodeIndexOffset) and that we always eventually try all nodes 
> in the face of connection failures (which isn't currently handled by 
> leastLoadedNode and probably cannot be without tracking additional state). 
> This method also has to work for new consumer use cases even though it is 
> currently only used by the new producer's metadata fetch. Finally it has to 
> properly handle when other code calls initiateConnect() since the normal path 
> for sending messages also initiates connections.
> We can already say that there is an order of preference given a single call 
> (as follows), but making this work across multiple calls when some initial 
> choices fail to connect or return metadata *and* connection states may be 
> changing is much more difficult.
>  * Connected, zero in flight requests - the request can be sent immediately
>  * Connecting node - it will hopefully be connected very soon and by 
> definition has no in flight requests
>  * Disconnected - same reasoning as for a connecting node
>  * Connected, > 0 in flight requests - we consider any # of in flight 
> requests as a big enough backlog to delay the request a lot.
> We could use an approach that better accounts for # of in flight requests 
> rather than just turning it into a boolean variable, but that probably 
> introduces much more complexity than it is worth.
> 3. The most difficult case to handle so far has been when leastLoadedNode 
> returns a disconnected node to maybeUpdateMetadata as its best option. 
> Properly handling the two resulting cases (initiateConnect fails immediately 
> vs. taking some time to possibly establish the connection) is tricky.
> 4. Consider optimizing for the failure cases. The most common cases are when 
> you already have an active connection and can immediately get the metadata or 
> you need to establish a connection, but the connection and metadata 
> request/response happen very quickly. These common cases are infrequent 
> enough (default every 5 min) that establishing an extra connection isn't a 
> big deal as long as it's eventually cleaned up. The edge cases, like network 
> partitions where some subset of nodes become unreachable for a long period, 
> are harder to reason about but we should be sure we will always be able to 
> gracefully recover from them.
> KAFKA-1642 enumerated the possible outcomes of a single 

[jira] [Issue Comment Deleted] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff updated KAFKA-3323:
--
Comment: was deleted

(was: Log Splitter utility.  For those with production log files that are 
affected by this issue, and cannot be dropped.  This utility will split 
existing log segments into legal chunks)

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1, 0.8.2.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Updated] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff updated KAFKA-3323:
--
Affects Version/s: 0.8.2.1
   Status: Patch Available  (was: Reopened)

Log Splitter utility.  For those with production log files that are affected by 
this issue, and cannot be dropped.  This utility will split existing log 
segments into legal chunks

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.2.1, 0.8.1.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Reopened] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff reopened KAFKA-3323:
---

Re-opening so that I can attach the log-splitter patch

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Comment Edited] (KAFKA-1843) Metadata fetch/refresh in new producer should handle all node connection states gracefully

[venu (JIRA)]

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

venu edited comment on KAFKA-1843 at 3/8/16 7:06 PM:
-

I am having similar problem , when partition is down, metadata is still 
returning the downed partitions as up. how to refresh/update the metadata when 
partition is down. Using Kafka 0.8.2


was (Author: venu516):
I am having similar problem , when partition is down, metadata is still 
returning the downed partitions as up. how to refresh/update the metadata when 
partition is down.

> Metadata fetch/refresh in new producer should handle all node connection 
> states gracefully
> --
>
> Key: KAFKA-1843
> URL: https://issues.apache.org/jira/browse/KAFKA-1843
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, producer 
>Affects Versions: 0.8.2.0
>Reporter: Ewen Cheslack-Postava
>Assignee: Eno Thereska
>Priority: Blocker
>  Labels: patch
> Fix For: 0.8.2.1
>
>
> KAFKA-1642 resolved some issues with the handling of broker connection states 
> to avoid high CPU usage, but made the minimal fix rather than the ideal one. 
> The code for handling the metadata fetch is difficult to get right because it 
> has to handle a lot of possible connectivity states and failure modes across 
> all the known nodes. It also needs to correctly integrate with the 
> surrounding event loop, providing correct poll() timeouts to both avoid busy 
> looping and make sure it wakes up and tries new nodes in the face of both 
> connection and request failures.
> A patch here should address a few issues:
> 1. Make sure connection timeouts, as implemented in KAFKA-1842, are cleanly 
> integrated. This mostly means that when a connecting node is selected to 
> fetch metadata from, that the code notices that and sets the next timeout 
> based on the connection timeout rather than some other backoff.
> 2. Rethink the logic and naming of NetworkClient.leastLoadedNode. That method 
> actually takes into account a) the current connectivity of each node, b) 
> whether the node had a recent connection failure, c) the "load" in terms of 
> in flight requests. It also needs to ensure that different clients don't use 
> the same ordering across multiple calls (which is already addressed in the 
> current code by nodeIndexOffset) and that we always eventually try all nodes 
> in the face of connection failures (which isn't currently handled by 
> leastLoadedNode and probably cannot be without tracking additional state). 
> This method also has to work for new consumer use cases even though it is 
> currently only used by the new producer's metadata fetch. Finally it has to 
> properly handle when other code calls initiateConnect() since the normal path 
> for sending messages also initiates connections.
> We can already say that there is an order of preference given a single call 
> (as follows), but making this work across multiple calls when some initial 
> choices fail to connect or return metadata *and* connection states may be 
> changing is much more difficult.
>  * Connected, zero in flight requests - the request can be sent immediately
>  * Connecting node - it will hopefully be connected very soon and by 
> definition has no in flight requests
>  * Disconnected - same reasoning as for a connecting node
>  * Connected, > 0 in flight requests - we consider any # of in flight 
> requests as a big enough backlog to delay the request a lot.
> We could use an approach that better accounts for # of in flight requests 
> rather than just turning it into a boolean variable, but that probably 
> introduces much more complexity than it is worth.
> 3. The most difficult case to handle so far has been when leastLoadedNode 
> returns a disconnected node to maybeUpdateMetadata as its best option. 
> Properly handling the two resulting cases (initiateConnect fails immediately 
> vs. taking some time to possibly establish the connection) is tricky.
> 4. Consider optimizing for the failure cases. The most common cases are when 
> you already have an active connection and can immediately get the metadata or 
> you need to establish a connection, but the connection and metadata 
> request/response happen very quickly. These common cases are infrequent 
> enough (default every 5 min) that establishing an extra connection isn't a 
> big deal as long as it's eventually cleaned up. The edge cases, like network 
> partitions where some subset of nodes become unreachable for a long period, 
> are harder to reason about but we should be sure we will always be able to 
> gracefully recover from them.
> KAFKA-1642 enumerated the possible outcomes of a single call to 
> maybeUpdateMetadata. 

[jira] [Updated] (KAFKA-3312) Add a offsets methods to ZkUtils and replace relevant usages

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian updated KAFKA-3312:
---
Affects Version/s: (was: 0.9.0.1)
   Status: Patch Available  (was: Open)

> Add a offsets methods to ZkUtils and replace relevant usages
> 
>
> Key: KAFKA-3312
> URL: https://issues.apache.org/jira/browse/KAFKA-3312
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Grant Henke
>Assignee: Vahid Hashemian
>
> There are many places in the code that manually build a zookeeper path and 
> get or update offsets. Moving this logic to a common location in ZkUtils 
> would be nice. 
> Ex:
> {code}
> zkUtils.readDataMaybeNull(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}")._1
> {code}
> {code}
>  zkUtils.readData(topicDirs.consumerOffsetDir + "/" + 
> topicAndPartition.partition)._1.toLong
> {code}
> {code}
> zkUtils.updatePersistentPath(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}",
>  partitionData.offset.toString)
> {code}



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

[jira] [Updated] (KAFKA-2273) Add rebalance with a minimal number of reassignments to server-defined strategy list

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian updated KAFKA-2273:
---
Status: Patch Available  (was: Reopened)

> Add rebalance with a minimal number of reassignments to server-defined 
> strategy list
> 
>
> Key: KAFKA-2273
> URL: https://issues.apache.org/jira/browse/KAFKA-2273
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Olof Johansson
>Assignee: Vahid Hashemian
>  Labels: newbie++, newbiee
> Fix For: 0.10.1.0
>
>
> Add a new partitions.assignment.strategy to the server-defined list that will 
> do reassignments based on moving as few partitions as possible. This should 
> be a quite common reassignment strategy especially for the cases where the 
> consumer has to maintain state, either in memory, or on disk.



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

[jira] [Reopened] (KAFKA-2273) Add rebalance with a minimal number of reassignments to server-defined strategy list

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian reopened KAFKA-2273:


> Add rebalance with a minimal number of reassignments to server-defined 
> strategy list
> 
>
> Key: KAFKA-2273
> URL: https://issues.apache.org/jira/browse/KAFKA-2273
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Olof Johansson
>Assignee: Vahid Hashemian
>  Labels: newbie++, newbiee
> Fix For: 0.10.1.0
>
>
> Add a new partitions.assignment.strategy to the server-defined list that will 
> do reassignments based on moving as few partitions as possible. This should 
> be a quite common reassignment strategy especially for the cases where the 
> consumer has to maintain state, either in memory, or on disk.



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

[jira] [Resolved] (KAFKA-2273) Add rebalance with a minimal number of reassignments to server-defined strategy list

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian resolved KAFKA-2273.

Resolution: Fixed

> Add rebalance with a minimal number of reassignments to server-defined 
> strategy list
> 
>
> Key: KAFKA-2273
> URL: https://issues.apache.org/jira/browse/KAFKA-2273
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Olof Johansson
>Assignee: Vahid Hashemian
>  Labels: newbie++, newbiee
> Fix For: 0.10.1.0
>
>
> Add a new partitions.assignment.strategy to the server-defined list that will 
> do reassignments based on moving as few partitions as possible. This should 
> be a quite common reassignment strategy especially for the cases where the 
> consumer has to maintain state, either in memory, or on disk.



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

[jira] [Commented] (KAFKA-1843) Metadata fetch/refresh in new producer should handle all node connection states gracefully

[venu (JIRA)]

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

venu commented on KAFKA-1843:
-

I am having similar problem , when partition is down, metadata is still 
returning the downed partitions as up. how to refresh/update the metadata when 
partition is down.

> Metadata fetch/refresh in new producer should handle all node connection 
> states gracefully
> --
>
> Key: KAFKA-1843
> URL: https://issues.apache.org/jira/browse/KAFKA-1843
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, producer 
>Affects Versions: 0.8.2.0
>Reporter: Ewen Cheslack-Postava
>Assignee: Eno Thereska
>Priority: Blocker
>  Labels: patch
> Fix For: 0.8.2.1
>
>
> KAFKA-1642 resolved some issues with the handling of broker connection states 
> to avoid high CPU usage, but made the minimal fix rather than the ideal one. 
> The code for handling the metadata fetch is difficult to get right because it 
> has to handle a lot of possible connectivity states and failure modes across 
> all the known nodes. It also needs to correctly integrate with the 
> surrounding event loop, providing correct poll() timeouts to both avoid busy 
> looping and make sure it wakes up and tries new nodes in the face of both 
> connection and request failures.
> A patch here should address a few issues:
> 1. Make sure connection timeouts, as implemented in KAFKA-1842, are cleanly 
> integrated. This mostly means that when a connecting node is selected to 
> fetch metadata from, that the code notices that and sets the next timeout 
> based on the connection timeout rather than some other backoff.
> 2. Rethink the logic and naming of NetworkClient.leastLoadedNode. That method 
> actually takes into account a) the current connectivity of each node, b) 
> whether the node had a recent connection failure, c) the "load" in terms of 
> in flight requests. It also needs to ensure that different clients don't use 
> the same ordering across multiple calls (which is already addressed in the 
> current code by nodeIndexOffset) and that we always eventually try all nodes 
> in the face of connection failures (which isn't currently handled by 
> leastLoadedNode and probably cannot be without tracking additional state). 
> This method also has to work for new consumer use cases even though it is 
> currently only used by the new producer's metadata fetch. Finally it has to 
> properly handle when other code calls initiateConnect() since the normal path 
> for sending messages also initiates connections.
> We can already say that there is an order of preference given a single call 
> (as follows), but making this work across multiple calls when some initial 
> choices fail to connect or return metadata *and* connection states may be 
> changing is much more difficult.
>  * Connected, zero in flight requests - the request can be sent immediately
>  * Connecting node - it will hopefully be connected very soon and by 
> definition has no in flight requests
>  * Disconnected - same reasoning as for a connecting node
>  * Connected, > 0 in flight requests - we consider any # of in flight 
> requests as a big enough backlog to delay the request a lot.
> We could use an approach that better accounts for # of in flight requests 
> rather than just turning it into a boolean variable, but that probably 
> introduces much more complexity than it is worth.
> 3. The most difficult case to handle so far has been when leastLoadedNode 
> returns a disconnected node to maybeUpdateMetadata as its best option. 
> Properly handling the two resulting cases (initiateConnect fails immediately 
> vs. taking some time to possibly establish the connection) is tricky.
> 4. Consider optimizing for the failure cases. The most common cases are when 
> you already have an active connection and can immediately get the metadata or 
> you need to establish a connection, but the connection and metadata 
> request/response happen very quickly. These common cases are infrequent 
> enough (default every 5 min) that establishing an extra connection isn't a 
> big deal as long as it's eventually cleaned up. The edge cases, like network 
> partitions where some subset of nodes become unreachable for a long period, 
> are harder to reason about but we should be sure we will always be able to 
> gracefully recover from them.
> KAFKA-1642 enumerated the possible outcomes of a single call to 
> maybeUpdateMetadata. A good fix for this would consider all of those outcomes 
> for repeated calls to 



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

[jira] [Updated] (KAFKA-2073) Replace TopicMetadata request/response with o.a.k.requests.metadata

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-2073:
---
Fix Version/s: 0.10.0.0

> Replace TopicMetadata request/response with o.a.k.requests.metadata
> ---
>
> Key: KAFKA-2073
> URL: https://issues.apache.org/jira/browse/KAFKA-2073
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Gwen Shapira
>Assignee: Jason Gustafson
> Fix For: 0.10.0.0
>
>
> Replace TopicMetadata request/response with o.a.k.requests.metadata.
> Note, this is more challenging that it appears because while the wire 
> protocol is identical, the objects are completely different.



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

[jira] [Resolved] (KAFKA-3032) Broken documentation link

[Grant Henke (JIRA)]

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

Grant Henke resolved KAFKA-3032.

Resolution: Fixed
  Assignee: Grant Henke

This was resolved via https://github.com/apache/kafka/pull/676

> Broken documentation link
> -
>
> Key: KAFKA-3032
> URL: https://issues.apache.org/jira/browse/KAFKA-3032
> Project: Kafka
>  Issue Type: Bug
>  Components: website
>Reporter: Vahid Hashemian
>Assignee: Grant Henke
>Priority: Minor
>  Labels: documentation
>
> The link http://kafka.apache.org/ops.html#quotas referenced in the 
> documentation on the website 
> (http://kafka.apache.org/documentation.html#design_quotasoverrides) is 
> broken. The file ops.html actually exists in the kafka project source, but 
> seems to be missing on the website.



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

[jira] [Resolved] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Ismael Juma (JIRA)]

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

Ismael Juma resolved KAFKA-3323.

Resolution: Duplicate

Thanks for confirming Michael. Closing as duplicate. Maybe you can attach a 
patch for the tool as others may also find it useful?

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Comment Edited] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff edited comment on KAFKA-3323 at 3/8/16 6:52 PM:
---

[~ijuma] Yes, it is a duplicate.  The accepted solution was the first one I 
mentioned in my comment: group segments to clean by segment size in bytes *and* 
by limiting the difference between the segment base offset and the max offset 
that goes into it.

This however does not address the already broken log segments that I (and 
others) may have in production.  For this I have written a utility to split log 
segments into legal chunks.  If this seems useful I can submit it as a patch.


was (Author: michael.schiff):
[~ijuma] Yes, it is a duplicate.  The accepted solution was the first one I 
mentioned in my comment: group segments to clean, by segment size in bytes and 
by limiting the difference between the segment base offset and the max offset 
that goes into it.

This however does not address the already broken log segments that I (and 
others) may have in production.  For this I have written a utility to split log 
segments into legal chunks.  If this seems useful I can submit it as a patch.

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Commented] (KAFKA-3323) Negative offsets in Log Segment Index files due to Integer overflow when compaction is enabled

[Michael Schiff (JIRA)]

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

Michael Schiff commented on KAFKA-3323:
---

[~ijuma] Yes, it is a duplicate.  The accepted solution was the first one I 
mentioned in my comment: group segments to clean, by segment size in bytes and 
by limiting the difference between the segment base offset and the max offset 
that goes into it.

This however does not address the already broken log segments that I (and 
others) may have in production.  For this I have written a utility to split log 
segments into legal chunks.  If this seems useful I can submit it as a patch.

> Negative offsets in Log Segment Index files due to Integer overflow when 
> compaction is enabled 
> ---
>
> Key: KAFKA-3323
> URL: https://issues.apache.org/jira/browse/KAFKA-3323
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.8.1.1
>Reporter: Michael Schiff
>Assignee: Jay Kreps
> Attachments: index_dump.txt, log_dump.txt
>
>
> Once the Offset Index has negative offset values, the binary search for 
> position lookup is broken. This causes consumers of compact topics to skip 
> large offset intervals when bootstrapping.  This has serious implications for 
> consumers of compact topics.
> {code}
>  /**
>* Append an entry for the given offset/location pair to the index. This 
> entry must have a larger offset than all subsequent entries.
>*/
>   def append(offset: Long, position: Int) {
> inLock(lock) {
>   require(!isFull, "Attempt to append to a full index (size = " + size + 
> ").")
>   if (size.get == 0 || offset > lastOffset) {
> debug("Adding index entry %d => %d to %s.".format(offset, position, 
> file.getName))
> this.mmap.putInt((offset - baseOffset).toInt)
> this.mmap.putInt(position)
> this.size.incrementAndGet()
> this.lastOffset = offset
> require(entries * 8 == mmap.position, entries + " entries but file 
> position in index is " + mmap.position + ".")
>   } else {
> throw new InvalidOffsetException("Attempt to append an offset (%d) to 
> position %d no larger than the last offset appended (%d) to %s."
>   .format(offset, entries, lastOffset, file.getAbsolutePath))
>   }
> }
>   }
> {code}
> OffsetIndex.append assumes that (offset - baseOffset) can be represented as 
> an integer without overflow. If the LogSegment is from a compacted topic, 
> this assumption may not be valid. The result is a quiet integer overflow, 
> which stores a negative value into the index.
> I believe that the issue is caused by the LogCleaner. Specifically, by the 
> groupings produced by 
> {code}
> /**
>* Group the segments in a log into groups totaling less than a given size. 
> the size is enforced separately for the log data and the index data.
>* We collect a group of such segments together into a single
>* destination segment. This prevents segment sizes from shrinking too much.
>*
>* @param segments The log segments to group
>* @param maxSize the maximum size in bytes for the total of all log data 
> in a group
>* @param maxIndexSize the maximum size in bytes for the total of all index 
> data in a group
>*
>* @return A list of grouped segments
>*/
>   private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], 
> maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]]
> {code}
> Since this method is only concerned with grouping by size, without taking 
> baseOffset and groupMaxOffset into account, it will produce groups that when 
> cleaned into a single segment, have offsets that overflow. This is more 
> likely for topics with low key cardinality, but high update volume, as you 
> could wind up with very few cleaned records, but with very high offsets.



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

[jira] [Updated] (KAFKA-3038) Speeding up partition reassignment after broker failure

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3038:
---
Fix Version/s: 0.11.0.0

> Speeding up partition reassignment after broker failure
> ---
>
> Key: KAFKA-3038
> URL: https://issues.apache.org/jira/browse/KAFKA-3038
> Project: Kafka
>  Issue Type: Improvement
>  Components: controller, core
>Affects Versions: 0.9.0.0
>Reporter: Eno Thereska
>Assignee: Eno Thereska
> Fix For: 0.11.0.0
>
>
> After a broker failure the controller does several writes to Zookeeper for 
> each partition on the failed broker. Writes are done one at a time, in closed 
> loop, which is slow especially under high latency networks. Zookeeper has 
> support for batching operations (the "multi" API). It is expected that 
> substituting serial writes with batched ones should reduce failure handling 
> time by an order of magnitude.
> This is identified as an issue in 
> https://cwiki.apache.org/confluence/display/KAFKA/kafka+Detailed+Replication+Design+V3
>  (section End-to-end latency during a broker failure)



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

[jira] [Updated] (KAFKA-3038) Speeding up partition reassignment after broker failure

[Grant Henke (JIRA)]

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

Grant Henke updated KAFKA-3038:
---
Fix Version/s: (was: 0.9.0.0)

> Speeding up partition reassignment after broker failure
> ---
>
> Key: KAFKA-3038
> URL: https://issues.apache.org/jira/browse/KAFKA-3038
> Project: Kafka
>  Issue Type: Improvement
>  Components: controller, core
>Affects Versions: 0.9.0.0
>Reporter: Eno Thereska
>Assignee: Eno Thereska
> Fix For: 0.11.0.0
>
>
> After a broker failure the controller does several writes to Zookeeper for 
> each partition on the failed broker. Writes are done one at a time, in closed 
> loop, which is slow especially under high latency networks. Zookeeper has 
> support for batching operations (the "multi" API). It is expected that 
> substituting serial writes with batched ones should reduce failure handling 
> time by an order of magnitude.
> This is identified as an issue in 
> https://cwiki.apache.org/confluence/display/KAFKA/kafka+Detailed+Replication+Design+V3
>  (section End-to-end latency during a broker failure)



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

[jira] [Commented] (KAFKA-3309) Update Protocol Documentation WIP patch

[Ashish K Singh (JIRA)]

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

Ashish K Singh commented on KAFKA-3309:
---

[~granthenke] it might need an update based on whether or not we add the 
deprecation list in Protocol. Based on recent development, it looks like we 
need a mechanism for deprecation, however that mechanism could be as simple as 
manually updating docs to reflect deprecation. Based on how discussion for 
KIP-35 goes, we can modify the intent or close the JIRA. Sounds reasonable?

> Update Protocol Documentation WIP patch
> ---
>
> Key: KAFKA-3309
> URL: https://issues.apache.org/jira/browse/KAFKA-3309
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
>
> https://github.com/apache/kafka/pull/970 has the WIP patch. Update to reflect 
> changes made for KIP-35.



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

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Gwen Shapira]

Yes, I understand that.

However, because the current configuration only allows a single
callback and a single login module, I need to supply a class that
supports callbacks or logins for every single mechanism that I need to
support. I question whether this is applicable in real-world scenario
where a user may need to support mechanisms that were written by
different vendors and where a single class that implements all the
necessary modules or callsbacks is very unlikely to exist.

I agree that supporting a list of classes and dynamically using the
correct callbacks and modules based on configuration is complex, but I
don't think this feature is useful enough without it.

Gwen

On Tue, Mar 8, 2016 at 2:40 AM, Rajini Sivaram
 wrote:
> Gwen,
>
> Sorry if this is not clear in the KIP. The goal of the KIP is to enable new
> mechanisms to be added without any changes to Apache Kafka. The
> configuration proposed by the KIP is based on the configurability of the
> Java SASL API and what it would take to add a new mechanism similar to
> Kerberos/PLAIN/Digest-MD5 and the requirements that Tao raised for custom
> authentication.
>
> The PR includes a test for Digest-MD5 which is not included in Kafka, but
> is tested using the configuration options that allow new mechanisms to be
> added.
>
> If you want to add a more complex mechanism which requires integration with
> some external authentication libraries for login or callback handlers, you
> can do that by configuring your own callback handler and login
> implementation. The login class allows you to add additional threads (for
> instance for token refresh), while the callback handler extends the
> standard Java CallbackHandler interface, enabling you to override any of
> the default callbacks in Kafka to provide your own behaviour. If you want
> to add your own login modules, SASL server or SASL client implementation
> for a mechanism that is not supported in your JDK, you can do that by
> adding new Java security modules, using the built-in pluggability of Java
> security.
>
> And finally, if you want to enable multiple mechanisms in your Kafka
> brokers, you can specify a callback handler class that either provides the
> handlers for all the mechanisms (as the PR does for Kerberos and PLAIN) or
> a delegator that calls different callback handlers based on the mechanism.
> And for login, you provide a single or a compound class that provides any
> additional login logic for all the login modules that you have specified
> for KafkaServer.
>
> Hope that helps.
>
>
> On Tue, Mar 8, 2016 at 1:17 AM, Gwen Shapira  wrote:
>
>> Can you explain the process for a adding a new mechanism based on current
>> KIP?
>>
>> My thought is that if it requires modifying Apache Kafka code, it is
>> not pluggable enough.
>>
>> On Mon, Mar 7, 2016 at 4:04 PM, Rajini Sivaram
>>  wrote:
>> > Gwen,
>> >
>> > In cases where you want completely different callbacks for different
>> > mechanisms, I was thinking that the choice would be between a map of
>> > classes (mechanism -> callbackHandler class) or a delegator class that
>> > chooses the appropriate callback handler class based on mechanism. I
>> chose
>> > the latter since it makes it easier to configure in Kafka. Since we
>> create
>> > a callback handler for each channel and configure it with the
>> > client-selected mechanism, it is straightforward to have one wrapper
>> class
>> > that delegates to the right mechanism-specific class to handle callbacks.
>> > In many cases, a single class may be sufficient (the PR uses a single
>> > callback class for Kerberos and PLAIN). I do see your point about the
>> > flexibility that multiple classes would provide, but since you need to be
>> > able to associate the callback with a mechanism for this to be useful, I
>> am
>> > not sure if just a list would add value.
>> >
>> > Login class is slightly different since the proposal is to use a single
>> > login context with multiple login modules to handle multiple mechanisms.
>> In
>> > this case, you want to perform login for all the mechanisms that are
>> > enabled. And you want to call loginContext.login() only once. Again, you
>> > can delegate to multiple classes if you wish to add some complex
>> > mechanism-specific logic, but a single login class makes the mapping to a
>> > single login context and the login cache more obvious (the PR has a test
>> > that includes Kerberos and PLAIN).
>> >
>> > Thoughts?
>> >
>> > On Mon, Mar 7, 2016 at 9:57 PM, Gwen Shapira  wrote:
>> >
>> >> Hi,
>> >>
>> >> Before I can vote on this KIP, I have two additional questions /
>> >> comments on the new configuration:
>> >>
>> >> 1. sasl.callback.handler.class - it looks like we want a single class
>> >> that implements all mechanisms. I think this will make it difficult to
>> >> extend since the only way I can add a mechanism will be by
>> >> implementing 

Re: [VOTE] Release plan - Kafka 0.10.0

[Grant Henke]

+1 (non-binding).

I will review and update any jiras I  think should be tracked today.

Gwen the release tracking page is awesome!
+1 (non-binding)

One clarification: there are currently 11 issues marked as blockers, is
that an accurate list?

http://bit.ly/21YCthZ

-Flavio

> On 08 Mar 2016, at 06:12, Harsha  wrote:
>
>
> +1
>
> Thanks,
> Harsha
> On Mon, Mar 7, 2016, at 09:49 PM, Jun Rao wrote:
>> +1
>>
>> Thanks,
>>
>> Jun
>>
>> On Mon, Mar 7, 2016 at 9:27 AM, Gwen Shapira  wrote:
>>
>>> Greetings Kafka Developer Community,
>>>
>>> As you all know, we have few big features that are almost complete
>>> (Timestamps! Interceptors! Streams!). It is time to start planning our
>>> next release.
>>>
>>> I suggest the following:
>>> * Cut branches on March 21st
>>> * Publish the first release candidate the next day
>>> * Start testing, finding important issues, fixing them, rolling out new
>>> releases
>>> * And eventually get a release candidate that we all agree is awesome
>>> enough to release. Hopefully this won't take too many iterations :)
>>>
>>> Note that this is a 2 weeks heads-up on branch cutting. After we cut
>>> branches, we will try to minimize cherrypicks to just critical bugs
>>> (because last major release was a bit insane).
>>> Therefore,  if you have a feature that you really want to see in
>>> 0.10.0 - you'll need to have it committed by March 21st. As a curtesy
>>> to the release manager, if you have features that you are not planning
>>> on getting in for 0.10.0, please change the "fix version" field in
>>> JIRA accordingly.
>>>
>>> I will send a heads-up few days before cutting branches, to give
>>> everyone a chance to get stragglers in.
>>>
>>> The vote will be open for 72 hours.
>>> All in favor, please reply with +1.
>>>
>>> Gwen Shapira
>>>