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

[Apache Jenkins Server]

See 

[jira] [Created] (KAFKA-4943) SCRAM secret's should be better protected with Zookeeper ACLs

[JIRA]

Johan Ström created KAFKA-4943:
--

 Summary: SCRAM secret's should be better protected with Zookeeper 
ACLs
 Key: KAFKA-4943
 URL: https://issues.apache.org/jira/browse/KAFKA-4943
 Project: Kafka
  Issue Type: Improvement
Reporter: Johan Ström


With the new SCRAM authenticator the secrets are stored in Zookeeper:
{code}
get /kafka/config/users/alice
{"version":1,"config":{"SCRAM-SHA-512":"salt=ODhnZjNkdWZibTV1cG1zdnV6bmh6djF3Mg==,stored_key=BAbHWHuGEb4m5+U+p0M9oFQmOPhU6M7q5jtZY8deDDoZCvxaqVNLz41yPzdgcp1WpiEBmfwYOuFlo9hMFKM7mA==,server_key=JW3KhpMeyUgh0OAC0kejuFUvUSlXBv/Z68tlfOWcMw5f5jrBwyBnjNQ9VZsSYz1AcI9IYaQ5S6H3yN39SieNiA==,iterations=4096"}}
{code}

These are stored without any ACL, and zookeeper-security-migration.sh does not 
seem to change that either:

{code}
getAcl /kafka/config/users/alice
'world,'anyone
: cdrwa

getAcl /kafka/config/users
'world,'anyone
: cdrwa

getAcl /kafka
'world,'anyone
: r
'sasl,'bob
: cdrwa

getAcl /kafka/config/changes
'world,'anyone
: r
'sasl,'bob
: cdrwa

{code}

The above output is after running security migrator, for some reason 
/kafka/config/users is ignored, but others are fixed..


Even if these where to be stored with secure ZkUtils#DefaultAcls, they would be 
world readable.

>From my (limited) point of view, they should be readable by Kafka only.





--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4943) SCRAM secret's should be better protected with Zookeeper ACLs

[JIRA]

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

Johan Ström commented on KAFKA-4943:


Seems the no-acl-set-at-all was fixed in 
3445ed0bab6a035c3bacf1fd5f2439e44fb3ca52 (in 0.10.2 branch), but I tested with 
0.10.2.0 release.

The issue with readable credentials is still valid though.

> SCRAM secret's should be better protected with Zookeeper ACLs
> -
>
> Key: KAFKA-4943
> URL: https://issues.apache.org/jira/browse/KAFKA-4943
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Johan Ström
>
> With the new SCRAM authenticator the secrets are stored in Zookeeper:
> {code}
> get /kafka/config/users/alice
> {"version":1,"config":{"SCRAM-SHA-512":"salt=ODhnZjNkdWZibTV1cG1zdnV6bmh6djF3Mg==,stored_key=BAbHWHuGEb4m5+U+p0M9oFQmOPhU6M7q5jtZY8deDDoZCvxaqVNLz41yPzdgcp1WpiEBmfwYOuFlo9hMFKM7mA==,server_key=JW3KhpMeyUgh0OAC0kejuFUvUSlXBv/Z68tlfOWcMw5f5jrBwyBnjNQ9VZsSYz1AcI9IYaQ5S6H3yN39SieNiA==,iterations=4096"}}
> {code}
> These are stored without any ACL, and zookeeper-security-migration.sh does 
> not seem to change that either:
> {code}
> getAcl /kafka/config/users/alice
> 'world,'anyone
> : cdrwa
> getAcl /kafka/config/users
> 'world,'anyone
> : cdrwa
> getAcl /kafka
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> getAcl /kafka/config/changes
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> {code}
> The above output is after running security migrator, for some reason 
> /kafka/config/users is ignored, but others are fixed..
> Even if these where to be stored with secure ZkUtils#DefaultAcls, they would 
> be world readable.
> From my (limited) point of view, they should be readable by Kafka only.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [VOTE] KIP-123: Allow per stream/table timestamp extractor

[Jeyhun Karimov]

Sorry for a super late update. I made an update on related PR.

Cheers,
Jeyhun

On Wed, Mar 22, 2017 at 9:09 PM Guozhang Wang  wrote:

> Jeyhun,
>
> Could you update the status of this KIP since it has been some time since
> the last vote?
>
> I'm +1 besides the minor comments mentioned above.
>
>
> Guozhang
>
>
> On Mon, Mar 6, 2017 at 3:14 PM, Jeyhun Karimov 
> wrote:
>
> > Sorry I was late. I will update javadocs in related methods to emphasize
> > that TimestampExtractor is stateless.
> >
> > Cheers,
> > Jeyhun
> >
> > On Mon, Mar 6, 2017 at 8:17 PM Guozhang Wang  wrote:
> >
> > > 1) Sounds good.
> > >
> > > 2) Yeah what I meant is to emphasize that TimestampExtractor to be
> > > stateless in the docs somewhere.
> > >
> > >
> > > Guozhang
> > >
> > >
> > > On Sun, Mar 5, 2017 at 4:27 PM, Matthias J. Sax  >
> > > wrote:
> > >
> > > > Guozhang,
> > > >
> > > > about renaming the config parameters. I like this idea, but want to
> > > > point out, that this change should be done in a backward compatible
> > way.
> > > > Thus, we need to keep (and only deprecate) the current parameter
> names.
> > > >
> > > > I am not sure about (2)? What do you worry about? Using a "stateful
> > > > extractor"? -- this would be an antipattern IMHO. We can clarify
> that a
> > > > TimestampExtrator should be stateless though (even if this should be
> > > > clear).
> > > >
> > > >
> > > > -Matthias
> > > >
> > > >
> > > > On 3/4/17 6:36 PM, Guozhang Wang wrote:
> > > > > Jeyhun,
> > > > >
> > > > > Thanks for proposing this KIP! And sorry for getting late in the
> > > > discussion.
> > > > >
> > > > > I have a general suggestion not directly related to this KIP and a
> > > couple
> > > > > of comments for this KIP here:
> > > > >
> > > > > I agree with Mathieu's observation, partly because we are now
> having
> > > lots
> > > > > of overloaded functions both in the DSL and in PAPI, and it would
> be
> > > > quite
> > > > > confusing to users. As Matthias mentioned we do have some plans to
> > > > refactor
> > > > > this API, but just wanted to point it out that this KIP may likely
> > urge
> > > > us
> > > > > to do the API refactoring sooner than planned. My personal
> preference
> > > > would
> > > > > be doing that the next release (i.e. 0.11.0.0 in June).
> > > > >
> > > > >
> > > > > Now some detailed comments:
> > > > >
> > > > > 1. I'd suggest change TIMESTAMP_EXTRACTOR_CLASS_CONFIG to
> > > > > "default.timestamp.extractor" or "global.timestamp.extractor" (also
> > the
> > > > > Java variable name can be changed accordingly) along with this
> > change.
> > > In
> > > > > addition, maybe we can piggy-backing this to also rename
> > > > > KEY_SERDE_CLASS_CONFIG/VALUE_SERDE_CLASS_CONFIG to "default.key.."
> > etc
> > > > in
> > > > > this KIP.
> > > > >
> > > > > 2. Another thing we should consider, is that since now we could
> > > > potentially
> > > > > use multiple timestamp extractor instances than a single one, this
> > may
> > > be
> > > > > breaking if user's customization did some global bookkeeping based
> on
> > > the
> > > > > previous assumption (maybe a wild thought but e.g. keeping track
> some
> > > > > global counts in the extractor as a local variable). We need to
> > clarify
> > > > > this change in the javadoc and also potentially in the upgrade web
> > doc
> > > > > sections.
> > > > >
> > > > >
> > > > >
> > > > > Guozhang
> > > > >
> > > > >
> > > > > On Wed, Mar 1, 2017 at 6:09 AM, Michael Noll  >
> > > > wrote:
> > > > >
> > > > >> +1 (non-binding)
> > > > >>
> > > > >> Thanks for the KIP!
> > > > >>
> > > > >> On Wed, Mar 1, 2017 at 1:49 PM, Bill Bejeck 
> > > wrote:
> > > > >>
> > > > >>> +1
> > > > >>>
> > > > >>> Thanks
> > > > >>> Bill
> > > > >>>
> > > > >>> On Wed, Mar 1, 2017 at 5:06 AM, Eno Thereska <
> > eno.there...@gmail.com
> > > >
> > > > >>> wrote:
> > > > >>>
> > > >  +1 (non binding).
> > > > 
> > > >  Thanks
> > > >  Eno
> > > > > On 28 Feb 2017, at 17:22, Matthias J. Sax <
> matth...@confluent.io
> > >
> > > > >>> wrote:
> > > > >
> > > > > +1
> > > > >
> > > > > Thanks a lot for the KIP!
> > > > >
> > > > > -Matthias
> > > > >
> > > > >
> > > > > On 2/28/17 1:35 AM, Damian Guy wrote:
> > > > >> Thanks for the KIP Jeyhun!
> > > > >>
> > > > >> +1
> > > > >>
> > > > >> On Tue, 28 Feb 2017 at 08:59 Jeyhun Karimov <
> > je.kari...@gmail.com
> > > >
> > > >  wrote:
> > > > >>
> > > > >>> Dear community,
> > > > >>>
> > > > >>> I'd like to start the vote for KIP-123:
> > > > >>> https://cwiki.apache.org/confluence/pages/viewpage.
> > > >  action?pageId=68714788
> > > > >>>
> > > > >>>
> > > > >>> Cheers,
> > > > >>> Jeyhun
> > > > >>> --
> > > > >>> -Cheers
> > > > >>>
> > > > >>> Jeyhun
> > > > >>>
> > > > >>
> > > > >
> > > > 
> > > > 
> > > > >>>
> > > > >>
> > > > >
> > > > >
> > > > >
> > > >
> > > >
> > >
> > >
> > > --
> >

[jira] [Commented] (KAFKA-4943) SCRAM secret's should be better protected with Zookeeper ACLs

[Ismael Juma (JIRA)]

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

Ismael Juma commented on KAFKA-4943:


cc [~rsivaram]

> SCRAM secret's should be better protected with Zookeeper ACLs
> -
>
> Key: KAFKA-4943
> URL: https://issues.apache.org/jira/browse/KAFKA-4943
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Johan Ström
>
> With the new SCRAM authenticator the secrets are stored in Zookeeper:
> {code}
> get /kafka/config/users/alice
> {"version":1,"config":{"SCRAM-SHA-512":"salt=ODhnZjNkdWZibTV1cG1zdnV6bmh6djF3Mg==,stored_key=BAbHWHuGEb4m5+U+p0M9oFQmOPhU6M7q5jtZY8deDDoZCvxaqVNLz41yPzdgcp1WpiEBmfwYOuFlo9hMFKM7mA==,server_key=JW3KhpMeyUgh0OAC0kejuFUvUSlXBv/Z68tlfOWcMw5f5jrBwyBnjNQ9VZsSYz1AcI9IYaQ5S6H3yN39SieNiA==,iterations=4096"}}
> {code}
> These are stored without any ACL, and zookeeper-security-migration.sh does 
> not seem to change that either:
> {code}
> getAcl /kafka/config/users/alice
> 'world,'anyone
> : cdrwa
> getAcl /kafka/config/users
> 'world,'anyone
> : cdrwa
> getAcl /kafka
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> getAcl /kafka/config/changes
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> {code}
> The above output is after running security migrator, for some reason 
> /kafka/config/users is ignored, but others are fixed..
> Even if these where to be stored with secure ZkUtils#DefaultAcls, they would 
> be world readable.
> From my (limited) point of view, they should be readable by Kafka only.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2726: MINOR: only log first exception in RecordCollector...

[dguy]

GitHub user dguy opened a pull request:

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

MINOR: only log first exception in RecordCollectorImpl producer callback



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

$ git pull https://github.com/dguy/kafka producer-send-exception

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

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


commit e94ecd53a236d0a0bf6c9e4b9950206c5be5f7b2
Author: Damian Guy 
Date:   2017-03-23T10:39:58Z

only log first exception in producer callback




---
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] [Work started] (KAFKA-4919) Document that stores must not be closed when Processors are closed

[Damian Guy (JIRA)]

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

Work on KAFKA-4919 started by Damian Guy.
-
> Document that stores must not be closed when Processors are closed
> --
>
> Key: KAFKA-4919
> URL: https://issues.apache.org/jira/browse/KAFKA-4919
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Elias Levy
>Assignee: Damian Guy
>
> I have a streams job, that previously worked, that consumes and writes to a 
> large number of topics with many partitions and that uses many threads.  I 
> upgraded the job to 0.10.2.0.  The job now fails after a short time running, 
> seemingly after a rebalance.
> {quote}
> WARN  2017-03-19 18:03:20,432 [StreamThread-10][StreamThread.java:160] : 
> Unexpected state transition from RUNNING to NOT_RUNNING
> {quote}
> The first observation is that Streams is no longer outputting exceptions and 
> backtraces.  I had to add code to get this information.
> The exception:
> {quote}
> Exception: org.apache.kafka.streams.errors.StreamsException: Exception caught 
> in process. taskId=1_225, processor=KSTREAM-SOURCE-03, 
> topic=some_topic, partition=225, offset=266411
> org.apache.kafka.streams.errors.StreamsException: Exception caught in 
> process. taskId=1_225, processor=KSTREAM-SOURCE-03, topic=some_topic, 
> partition=225, offset=266411
>   at 
> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:641)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368)
> Caused by: org.apache.kafka.streams.errors.InvalidStateStoreException: Store 
> someStore-201701060400 is currently closed
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.validateStoreOpen(RocksDBStore.java:205)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.put(RocksDBStore.java:221)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.put(RocksDBSegmentedBytesStore.java:74)
>   at 
> org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.put(ChangeLoggingSegmentedBytesStore.java:54)
>   at 
> org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.put(MeteredSegmentedBytesStore.java:101)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:109)
>   ... X more
> {quote}
> The error occurs for many partitions.
> This was preceded by (for this partition):
> {quote}
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][ConsumerCoordinator.java:393] 
> : Revoking previously assigned partitions [some_topic-225] for group some_job
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:254] : 
> stream-thread [StreamThread-10] partitions [[some_topic-225]] revoked at the 
> beginning of consumer rebalance.
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:1056] : 
> stream-thread [StreamThread-10] Closing a task's topology 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:534] : 
> stream-thread [StreamThread-10] Committing consumer offsets of task 1_225
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1012] : 
> stream-thread [StreamThread-10] Updating suspended tasks to contain active 
> tasks [[1_225, 0_445, 0_30]]
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1019] : 
> stream-thread [StreamThread-10] Removing all active tasks [[1_225, 0_445, 
> 0_30]]
> INFO  2017-03-19 18:03:19,925 [StreamThread-10][ConsumerCoordinator.java:252] 
> : Setting newly assigned partitions [some_tpoic-225] for group some_job
> INFO  2017-03-19 18:03:19,927 [StreamThread-10][StreamThread.java:228] : 
> stream-thread [StreamThread-10] New partitions [[some_topic-225]] assigned at 
> the end of consumer rebalance.
> INFO  2017-03-19 18:03:19,929 [StreamThread-10][StreamTask.java:333] : task 
> [1_225] Initializing processor nodes of the topology
> Something happens.  What ???
> INFO  2017-03-19 18:03:20,135 [StreamThread-10][StreamThread.java:1045] : 
> stream-thread [StreamThread-10] Closing a task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:523] : 
> stream-thread [StreamThread-10] Closing the state manager of task 1_225
> INFO  2017-03-19 18:03:20,432 [StreamThread-10][StreamThread.j

[jira] [Assigned] (KAFKA-4919) Document that stores must not be closed when Processors are closed

[Damian Guy (JIRA)]

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

Damian Guy reassigned KAFKA-4919:
-

Assignee: Damian Guy

> Document that stores must not be closed when Processors are closed
> --
>
> Key: KAFKA-4919
> URL: https://issues.apache.org/jira/browse/KAFKA-4919
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Elias Levy
>Assignee: Damian Guy
>
> I have a streams job, that previously worked, that consumes and writes to a 
> large number of topics with many partitions and that uses many threads.  I 
> upgraded the job to 0.10.2.0.  The job now fails after a short time running, 
> seemingly after a rebalance.
> {quote}
> WARN  2017-03-19 18:03:20,432 [StreamThread-10][StreamThread.java:160] : 
> Unexpected state transition from RUNNING to NOT_RUNNING
> {quote}
> The first observation is that Streams is no longer outputting exceptions and 
> backtraces.  I had to add code to get this information.
> The exception:
> {quote}
> Exception: org.apache.kafka.streams.errors.StreamsException: Exception caught 
> in process. taskId=1_225, processor=KSTREAM-SOURCE-03, 
> topic=some_topic, partition=225, offset=266411
> org.apache.kafka.streams.errors.StreamsException: Exception caught in 
> process. taskId=1_225, processor=KSTREAM-SOURCE-03, topic=some_topic, 
> partition=225, offset=266411
>   at 
> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:641)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368)
> Caused by: org.apache.kafka.streams.errors.InvalidStateStoreException: Store 
> someStore-201701060400 is currently closed
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.validateStoreOpen(RocksDBStore.java:205)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.put(RocksDBStore.java:221)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.put(RocksDBSegmentedBytesStore.java:74)
>   at 
> org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.put(ChangeLoggingSegmentedBytesStore.java:54)
>   at 
> org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.put(MeteredSegmentedBytesStore.java:101)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:109)
>   ... X more
> {quote}
> The error occurs for many partitions.
> This was preceded by (for this partition):
> {quote}
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][ConsumerCoordinator.java:393] 
> : Revoking previously assigned partitions [some_topic-225] for group some_job
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:254] : 
> stream-thread [StreamThread-10] partitions [[some_topic-225]] revoked at the 
> beginning of consumer rebalance.
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:1056] : 
> stream-thread [StreamThread-10] Closing a task's topology 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:534] : 
> stream-thread [StreamThread-10] Committing consumer offsets of task 1_225
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1012] : 
> stream-thread [StreamThread-10] Updating suspended tasks to contain active 
> tasks [[1_225, 0_445, 0_30]]
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1019] : 
> stream-thread [StreamThread-10] Removing all active tasks [[1_225, 0_445, 
> 0_30]]
> INFO  2017-03-19 18:03:19,925 [StreamThread-10][ConsumerCoordinator.java:252] 
> : Setting newly assigned partitions [some_tpoic-225] for group some_job
> INFO  2017-03-19 18:03:19,927 [StreamThread-10][StreamThread.java:228] : 
> stream-thread [StreamThread-10] New partitions [[some_topic-225]] assigned at 
> the end of consumer rebalance.
> INFO  2017-03-19 18:03:19,929 [StreamThread-10][StreamTask.java:333] : task 
> [1_225] Initializing processor nodes of the topology
> Something happens.  What ???
> INFO  2017-03-19 18:03:20,135 [StreamThread-10][StreamThread.java:1045] : 
> stream-thread [StreamThread-10] Closing a task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:523] : 
> stream-thread [StreamThread-10] Closing the state manager of task 1_225
> INFO  2017-03-19 18:03:20,432 [StreamThread-10][Str

[DISCUSS] KIP-134: Delay initial consumer group rebalance

[Damian Guy]

Hi All,

I've prepared a KIP to add a configurable delay to the initial consumer
group rebalance.

Please have look here:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-134%3A+Delay+initial+consumer+group+rebalance

Thanks,
Damian

BTW, i apologize if this appears twice. Seems the first one may have not
made it.

Re: [DISCUSS] KIP-134: Delay initial consumer group rebalance

[Ismael Juma]

Thanks for the KIP, Damian. It makes sense to avoid multiple rebalances
during start-up. One issue with having this as a broker config is that it
may be difficult to choose the right delay for all consumer groups. Can you
elaborate a little more on why the first alternative (add a consumer
config) was rejected? We bump protocol versions regularly (when it makes
sense), so it would be good to get a bit more detail.

Thanks,
Ismael

On Thu, Mar 23, 2017 at 12:24 PM, Damian Guy  wrote:

> Hi All,
>
> I've prepared a KIP to add a configurable delay to the initial consumer
> group rebalance.
>
> Please have look here:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 134%3A+Delay+initial+consumer+group+rebalance
>
> Thanks,
> Damian
>
> BTW, i apologize if this appears twice. Seems the first one may have not
> made it.
>

Re: [DISCUSS] KIP-134: Delay initial consumer group rebalance

[Damian Guy]

Hi Ismael,

Mostly to avoid the protocol bump.

I agree that it may be difficult to choose the right delay for all consumer
groups, but we wanted to make this something that most users don't really
need to think about, i.e., a small enough default delay that works in the
majority of cases. However it would be much more flexible as a consumer
config, which i'm happy to pursue if this change is worthy of a protocol
bump.

Thanks,
Damian

On Thu, 23 Mar 2017 at 12:35 Ismael Juma  wrote:

> Thanks for the KIP, Damian. It makes sense to avoid multiple rebalances
> during start-up. One issue with having this as a broker config is that it
> may be difficult to choose the right delay for all consumer groups. Can you
> elaborate a little more on why the first alternative (add a consumer
> config) was rejected? We bump protocol versions regularly (when it makes
> sense), so it would be good to get a bit more detail.
>
> Thanks,
> Ismael
>
> On Thu, Mar 23, 2017 at 12:24 PM, Damian Guy  wrote:
>
> > Hi All,
> >
> > I've prepared a KIP to add a configurable delay to the initial consumer
> > group rebalance.
> >
> > Please have look here:
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > 134%3A+Delay+initial+consumer+group+rebalance
> >
> > Thanks,
> > Damian
> >
> > BTW, i apologize if this appears twice. Seems the first one may have not
> > made it.
> >
>

[jira] [Commented] (KAFKA-4930) Connect Rest API allows creating connectors with an empty name

[JIRA]

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

Sönke Liebau commented on KAFKA-4930:
-

After a good nights sleep I am beginning to think that between this and 
[KAFKA-4938|https://issues.apache.org/jira/browse/KAFKA-4938] a better approach 
might be to change the _nonEmpty_ Validator to a more specialised _ValidName_ 
validator, which could check for empty strings, backslashes and other 
problematic characters (I think there are a few more jiras around).
The problematic check in 
[KAFKA-4938|https://issues.apache.org/jira/browse/KAFKA-4938] could be replaced 
by a check for null, if no name parameter is present in the request at all, 
which we could consider a bad request I think, so the BadRequest Exception 
makes sense here.

> Connect Rest API allows creating connectors with an empty name
> --
>
> Key: KAFKA-4930
> URL: https://issues.apache.org/jira/browse/KAFKA-4930
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Sönke Liebau
>Priority: Minor
>
> The Connect Rest API allows to deploy connectors with an empty name field, 
> which then cannot be removed through the api.
> Sending the following request:
> {code}
> {
> "name": "",
> "config": {
> "connector.class": 
> "org.apache.kafka.connect.tools.MockSourceConnector",
> "tasks.max": "1",
> "topics": "test-topic"
> 
> }
> }
> {code}
> Results in a connector being deployed which can be seen in the list of 
> connectors:
> {code}
> [
>   "",
>   "testconnector"
> ]{code}
> But cannot be removed via a DELETE call, as the api thinks we are trying to 
> delete the /connectors endpoint and declines the request.
> I don't think there is a valid case for the connector name to be empty so 
> perhaps we should add a check for this. I am happy to work on this.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

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

[Apache Jenkins Server]

See 

--
[...truncated 710.12 KB...]

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > testRestore 
STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > testRestore 
PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRange STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRange PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRangeWithDefaultSerdes STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRangeWithDefaultSerdes PASSED

org.apache.kafka.streams.KeyValueTest > shouldHaveSaneEqualsAndHashCode STARTED

org.apache.kafka.streams.KeyValueTest > shouldHaveSaneEqualsAndHashCode PASSED

org.apache.kafka.streams.KafkaStreamsTest > testIllegalMetricsConfig STARTED

org.apache.kafka.streams.KafkaStreamsTest > testIllegalMetricsConfig PASSED

org.apache.kafka.streams.KafkaStreamsTest > shouldNotGetAllTasksWhenNotRunning 
STARTED

org.apache.kafka.streams.KafkaStreamsTest > shouldNotGetAllTasksWhenNotRunning 
PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
testInitializesAndDestroysMetricsReporters STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
testInitializesAndDestroysMetricsReporters PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndPartitionerWhenNotRunning STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndPartitionerWhenNotRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > testLegalMetricsConfig STARTED

org.apache.kafka.streams.KafkaStreamsTest > testLegalMetricsConfig PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndSerializerWhenNotRunning STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndSerializerWhenNotRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldReturnFalseOnCloseWhenThreadsHaventTerminated STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldReturnFalseOnCloseWhenThreadsHaventTerminated PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetAllTasksWithStoreWhenNotRunning STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetAllTasksWithStoreWhenNotRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartOnceClosed STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartOnceClosed PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCleanup STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCleanup PASSED

org.apache.kafka.streams.KafkaStreamsTest > testNumberDefaultMetrics STARTED

org.apache.kafka.streams.KafkaStreamsTest > testNumberDefaultMetrics PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCloseIsIdempotent STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCloseIsIdempotent PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCannotCleanupWhileRunning 
STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCannotCleanupWhileRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartTwice STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartTwice PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableJoin PASSED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowExceptionOverlappingPattern STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowExceptionOverlappingPattern PASSED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowExceptionOverlappingTopic STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowExceptionOverlappingTopic PASSED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowStreamsExceptionNoResetSpecified STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowStreamsExceptionNoResetSpecified PASSED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldOnlyReadRecordsWhereEarliestSpecified STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldOnlyReadRecordsWhereEarliestSpecified PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldNotMakeStoreAvailableUntilAllStoresAvailable STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest

[jira] [Created] (KAFKA-4944) Fix an "unread field" findbugs warning in streams examples

[Colin P. McCabe (JIRA)]

Colin P. McCabe created KAFKA-4944:
--

 Summary: Fix an "unread field" findbugs warning in streams examples
 Key: KAFKA-4944
 URL: https://issues.apache.org/jira/browse/KAFKA-4944
 Project: Kafka
  Issue Type: Bug
Reporter: Colin P. McCabe
Assignee: Colin P. McCabe
Priority: Minor


Fix an "unread field" findbugs warning in streams examples



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2727: KAFKA-4944. Fix an "unread field" findbugs warning...

[cmccabe]

GitHub user cmccabe opened a pull request:

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

KAFKA-4944. Fix an "unread field" findbugs warning in streams examples 



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

$ git pull https://github.com/cmccabe/kafka KAFKA-4944

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

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






---
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-4944) Fix an "unread field" findbugs warning in streams examples

[ASF GitHub Bot (JIRA)]

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

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

GitHub user cmccabe opened a pull request:

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

KAFKA-4944. Fix an "unread field" findbugs warning in streams examples 



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

$ git pull https://github.com/cmccabe/kafka KAFKA-4944

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

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






> Fix an "unread field" findbugs warning in streams examples
> --
>
> Key: KAFKA-4944
> URL: https://issues.apache.org/jira/browse/KAFKA-4944
> Project: Kafka
>  Issue Type: Bug
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
>Priority: Minor
>
> Fix an "unread field" findbugs warning in streams examples



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [VOTE] KIP-122: Add Reset Consumer Group Offsets tooling

[Mickael Maison]

+1 (non binding)
Thanks for the KIP

On Wed, Mar 22, 2017 at 11:00 AM, Jorge Esteban Quilcate Otoya
 wrote:
> @Jason, thanks for your feedback!
> You're right, we are not considering the old consumer, since we rely on the
> KafkaConsumer#seek operations. I'm happy to update the KIP to make this
> explicit.
> About the second comment: I suppose that it would work, but I would like to
> include it to the test cases first. Do you know if this scenario has been
> test it in other clients?
>
> Jorge
>
>
> El mié., 22 mar. 2017 a las 5:23, Dong Lin () escribió:
>
>> Thanks for the KIP!
>>
>> +1 (non-binding)
>>
>> On Tue, Mar 21, 2017 at 6:24 PM, Becket Qin  wrote:
>>
>> > +1
>> >
>> > Thanks for the KIP. The tool is very useful.
>> >
>> > On Tue, Mar 21, 2017 at 4:46 PM, Jason Gustafson 
>> > wrote:
>> >
>> > > +1 This looks super useful! Might be worth mentioning somewhere
>> > > compatibility with the old consumer. It looks like offsets in zk are
>> not
>> > > covered, which seems fine, but probably should be explicitly noted.
>> Maybe
>> > > you can also add a note saying that the tool can be used for old
>> > consumers
>> > > which have offsets stored in Kafka, but it will not protect against an
>> > > active consumer group in that case?
>> > >
>> > > Thanks,
>> > > Jason
>> > >
>> > > On Tue, Mar 14, 2017 at 10:13 AM, Dong Lin 
>> wrote:
>> > >
>> > > > +1 (non-binding)
>> > > >
>> > > > On Tue, Mar 14, 2017 at 8:53 AM, Bill Bejeck 
>> > wrote:
>> > > >
>> > > > > +1
>> > > > >
>> > > > > On Tue, Mar 14, 2017 at 11:50 AM, Grant Henke > >
>> > > > wrote:
>> > > > >
>> > > > > > +1. Agreed. This is a great tool to have.
>> > > > > >
>> > > > > > On Tue, Mar 14, 2017 at 12:33 AM, Gwen Shapira <
>> g...@confluent.io>
>> > > > > wrote:
>> > > > > >
>> > > > > > > +1 (binding)
>> > > > > > >
>> > > > > > > Nice job - this is going to be super useful.
>> > > > > > >
>> > > > > > > On Thu, Feb 23, 2017 at 4:46 PM, Jorge Esteban Quilcate Otoya <
>> > > > > > > quilcate.jo...@gmail.com> wrote:
>> > > > > > >
>> > > > > > > > Hi All,
>> > > > > > > >
>> > > > > > > > It seems that there is no further concern with the KIP-122.
>> > > > > > > > At this point we would like to start the voting process.
>> > > > > > > >
>> > > > > > > > The KIP can be found here:
>> > > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
>> > > > > > > > 122%3A+Add+Reset+Consumer+Group+Offsets+tooling
>> > > > > > > >
>> > > > > > > >
>> > > > > > > > Thanks!
>> > > > > > > >
>> > > > > > > > Jorge.
>> > > > > > > >
>> > > > > > >
>> > > > > > >
>> > > > > > >
>> > > > > > > --
>> > > > > > > *Gwen Shapira*
>> > > > > > > Product Manager | Confluent
>> > > > > > > 650.450.2760 <(650)%20450-2760> | @gwenshap
>> > > > > > > Follow us: Twitter  | blog
>> > > > > > > 
>> > > > > > >
>> > > > > >
>> > > > > >
>> > > > > >
>> > > > > > --
>> > > > > > Grant Henke
>> > > > > > Software Engineer | Cloudera
>> > > > > > gr...@cloudera.com | twitter.com/gchenke |
>> > > linkedin.com/in/granthenke
>> > > > > >
>> > > > >
>> > > >
>> > >
>> >
>>

Kafka queue full configuration

[Mohapatra, Sudhir (Nokia - IN/Gurgaon)]

Hi,
We are trying to simulate the kafka queue full scenarios on kafka 0.10.0.
I have seen in earlier versions there is a configuration parameter 
"queue.buffering.max.messages" which can be set to simulate the queue full 
scenario.
But in the kafka 0.10.0 this parameter is not there.
https://kafka.apache.org/0100/documentation.html

Is this "queue.buffering.max.messages" config parameter has been changed in 
kafka 0.10.0 release?
Can you please let me know what is the new parameter name for the same 
functionality?

Regards,
Sudhir

Re: [VOTE] KIP-122: Add Reset Consumer Group Offsets tooling

[Ismael Juma]

Hi Jorge,

Thanks for the detailed KIP. The tool looks very useful. A few comments:

1. We are using the default timezone of the client for the specified date.
This seems a bit error prone. Would it be better to require the users to
specify the time zone as part of the date time? We should at least allow
it, but my experience when it comes to using the default time zone in a
distributed environment is not great.
2. It seems like we are using the ISO 8601 format for date time and
duration. It would be good to mention that.
3. `shift-by` should perhaps be `shift-offset-by` to be a bit clearer.
4. Printing the current offset via reset-offsets is a bit odd, can we not
use a different option for that?
5. It's a bit odd that `by-duration` subtracts while `shift-by` moves
forward. It would be nice if the name made it clear that `by-duration` is
subtracting, but I have no good suggestions, so maybe that's the best we
can do.

Ismael

On Fri, Feb 24, 2017 at 12:46 AM, Jorge Esteban Quilcate Otoya <
quilcate.jo...@gmail.com> wrote:

> Hi All,
>
> It seems that there is no further concern with the KIP-122.
> At this point we would like to start the voting process.
>
> The KIP can be found here:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 122%3A+Add+Reset+Consumer+Group+Offsets+tooling
>
>
> Thanks!
>
> Jorge.
>

[jira] [Commented] (KAFKA-4159) Allow overriding producer & consumer properties at the connector level

[Stephen Durfey (JIRA)]

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

Stephen Durfey commented on KAFKA-4159:
---

So, full disclosure, I believe my original desire for overriding the group.id 
is alleviated once I can upgrade to kafka 0.10.2.0 and use single message 
transforms. 

The current use case for me was pulling data from a sql server database over 
jdbc to push into kafka. In the platform the data is organized into topics with 
a source identifier in the topic name with our own wrapper avro for the payload 
to provide additional metadata about the data coming from the database. Because 
the data needed to be wrapped I wasn't able to write directly to these topics 
with kafka connect, so they needed to be written to different topics. I had 
additional connectors pulling from those topics to transform the data into the 
format I needed and had a custom sink connector write out the wrapped paylods. 
To generate the metadata it was specific to a table, so the design had a 
transformation connector per topic that kafka connect wrote to. 

I wanted to be able to provide progress of transforming this data through the 
[kafka offset monitor|https://github.com/quantifind/KafkaOffsetMonitor], but 
having a connector per topic with a unique group.id per connector really made 
the offset monitor almost useless, as there were just way too many consumer 
groups, and as we scaled for more databases, it was only going to get worse. 
Since all the tables belonged to one source, I wanted to be able to group all 
the topics for that one source into a single identifier. That way I could just 
click on that identifier in the offset monitor and view the progress for all 
topics for that particular source. 

I could've overridden the group.id at the worker level and that would work as 
well, as long as the connect instance was only ever going to be for that 
source. However, I think long term that wasn't going to be the case. 

> Allow overriding producer & consumer properties at the connector level
> --
>
> Key: KAFKA-4159
> URL: https://issues.apache.org/jira/browse/KAFKA-4159
> Project: Kafka
>  Issue Type: New Feature
>  Components: KafkaConnect
>Reporter: Shikhar Bhushan
>Assignee: Stephen Durfey
>
> As an example use cases, overriding a sink connector's consumer's partition 
> assignment strategy.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [VOTE] KIP-82 Add Record Headers

[Michael Pearce]

Thanks all for the +1 votes so far, just one more committer vote needed (

Please note:

I have made one small adjustment to the kip based on Ismael’s comment in 
discussion thread, and further requested by Jason in the vote thread.

Please note the below method is changed based on this feedback.

Headers.header(key)` to be `Headers.lastHeader(key)

Thanks
Mike

On 22/03/2017, 16:39, "Joel Koshy"  wrote:

+1

On Tue, Mar 21, 2017 at 5:01 PM, Jason Gustafson  wrote:

> Thanks for the KIP! +1 (binding) from me. Just one nit: can we change
> `Headers.header(key)` to be `Headers.lastHeader(key)`? It's not a
> deal-breaker, but I think it's better to let the name reflect the actual
> behavior as clearly as possible.
>
> -Jason
>
> On Wed, Feb 15, 2017 at 6:10 AM, Jeroen van Disseldorp 
> wrote:
>
> > +1 on introducing the concept of headers, neutral on specific
> > implementation.
> >
> >
> >
> > On 14/02/2017 22:34, Jay Kreps wrote:
> >
> >> Couple of things I think we still need to work out:
> >>
> >> 1. I think we agree about the key, but I think we haven't talked
> about
> >> the value yet. I think if our goal is an open ecosystem of these
> >> header
> >> spread across many plugins from many systems we should consider
> >> making this
> >> a string as well so it can be printed, set via a UI, set in config,
> >> etc.
> >> Basically encouraging pluggable serialization formats here will 
lead
> >> to a
> >> bit of a tower of babel.
> >> 2. This proposal still includes a pretty big change to our
> >> serialization
> >> and protocol definition layer. Essentially it is introducing an
> >> optional
> >> type, where the format is data dependent. I think this is actually 
a
> >> big
> >> change though it doesn't seem like it. It means you can no longer
> >> specify
> >> this type with our type definition DSL, and likewise it requires
> >> custom
> >> handling in client libs. This isn't a huge thing, since the Record
> >> definition is custom anyway, but I think this kind of protocol
> >> inconsistency is very non-desirable and ties you to hand-coding
> >> things. I
> >> think the type should instead by [Key Value] in our BNF, where key
> and
> >> value are both short strings as used elsewhere. This brings it in
> >> line with
> >> the rest of the protocol.
> >> 3. Could we get more specific about the exact Java API change to
> >> ProducerRecord, ConsumerRecord, Record, etc?
> >>
> >> -Jay
> >>
> >> On Tue, Feb 14, 2017 at 9:42 AM, Michael Pearce 
> >> wrote:
> >>
> >> Hi all,
> >>>
> >>> We would like to start the voting process for KIP-82 – Add record
> >>> headers.
> >>> The KIP can be found
> >>> at
> >>>
> >>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> >>> 82+-+Add+Record+Headers
> >>>
> >>> Discussion thread(s) can be found here:
> >>>
> >>> http://search-hadoop.com/m/Kafka/uyzND1nSTOHTvj81?subj=
> >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> >>> http://search-hadoop.com/m/Kafka/uyzND1Arxt22Tvj81?subj=
> >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> >>> http://search-hadoop.com/?project=Kafka&q=KIP-82
> >>>
> >>>
> >>>
> >>> Thanks,
> >>> Mike
> >>>
> >>> The information contained in this email is strictly confidential and
> for
> >>> the use of the addressee only, unless otherwise indicated. If you are
> not
> >>> the intended recipient, please do not read, copy, use or disclose to
> >>> others
> >>> this message or any attachment. Please also notify the sender by
> replying
> >>> to this email or by telephone (+44(020 7896 0011) and then delete the
> >>> email
> >>> and any copies of it. Opinions, conclusion (etc) that do not relate to
> >>> the
> >>> official business of this company shall be understood as neither given
> >>> nor
> >>> endorsed by it. IG is a trading name of IG Markets Limited (a company
> >>> registered in England and Wales, company number 04008957) and IG Index
> >>> Limited (a company registered in England and Wales, company number
> >>> 01190902). Registered address at Cannon Bridge House, 25 Dowgate Hill,
> >>> London EC4R 2YA. Both IG Markets Limited (register number 195355) and
> IG
> >>> Index Limited (register number 114059) are authorised and regulated by
> >>> the
> >>> Financial Conduct Authority.
> >>>
> >>>
> >
>


The information contained in this email is strictly confidential and for the 
use of the addressee only, unless otherwise indicated. If you are not the 
intended recipient, please do not read, copy, use or disclose to others this 
message or any attachmen

[jira] [Commented] (KAFKA-4943) SCRAM secret's should be better protected with Zookeeper ACLs

[Rajini Sivaram (JIRA)]

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

Rajini Sivaram commented on KAFKA-4943:
---

The current SCRAM implementation uses the default ACL settings for Kafka with 
the expectation that SCRAM credentials are stored in Zookeeper only in 
installations where Zookeeper is safe (secure disk, network segmentation to 
restrict access etc.). Since ZK traffic is not encrypted, is would be unsafe to 
use ZK as the credential store in clusters where ZK is not fully secure. KIP-86 
(https://cwiki.apache.org/confluence/display/KAFKA/KIP-86%3A+Configurable+SASL+callback+handlers)
 will enable a pluggable credential store for environments where ZK is 
insecure. Having said that, we could probably do better for the default case as 
suggested here.

> SCRAM secret's should be better protected with Zookeeper ACLs
> -
>
> Key: KAFKA-4943
> URL: https://issues.apache.org/jira/browse/KAFKA-4943
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Johan Ström
>
> With the new SCRAM authenticator the secrets are stored in Zookeeper:
> {code}
> get /kafka/config/users/alice
> {"version":1,"config":{"SCRAM-SHA-512":"salt=ODhnZjNkdWZibTV1cG1zdnV6bmh6djF3Mg==,stored_key=BAbHWHuGEb4m5+U+p0M9oFQmOPhU6M7q5jtZY8deDDoZCvxaqVNLz41yPzdgcp1WpiEBmfwYOuFlo9hMFKM7mA==,server_key=JW3KhpMeyUgh0OAC0kejuFUvUSlXBv/Z68tlfOWcMw5f5jrBwyBnjNQ9VZsSYz1AcI9IYaQ5S6H3yN39SieNiA==,iterations=4096"}}
> {code}
> These are stored without any ACL, and zookeeper-security-migration.sh does 
> not seem to change that either:
> {code}
> getAcl /kafka/config/users/alice
> 'world,'anyone
> : cdrwa
> getAcl /kafka/config/users
> 'world,'anyone
> : cdrwa
> getAcl /kafka
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> getAcl /kafka/config/changes
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> {code}
> The above output is after running security migrator, for some reason 
> /kafka/config/users is ignored, but others are fixed..
> Even if these where to be stored with secure ZkUtils#DefaultAcls, they would 
> be world readable.
> From my (limited) point of view, they should be readable by Kafka only.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2728: KAFKA 3514: alternate calculation for min timestam...

[enothereska]

GitHub user enothereska opened a pull request:

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

KAFKA 3514: alternate calculation for min timestamp [WiP]

This alternate calculation reduces the amount of buffering kept and is 
optimised for cases when out of order records are rare. 

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

$ git pull https://github.com/enothereska/kafka 
KAFKA-3514-stream-timestamp-min

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

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


commit 1d1f7f79f396812c51515de50c18087d120eb092
Author: Eno Thereska 
Date:   2017-03-02T10:43:47Z

Initial poc

commit 1fcb484d5a294bbf289dca0735aaaccff93114ca
Author: Eno Thereska 
Date:   2017-03-02T14:48:58Z

Checkpoint test code

commit 28eb75185546f5947beda176c3e339ec9311
Author: Eno Thereska 
Date:   2017-03-05T08:55:15Z

Checkpoint

commit 4c68bf1f90fcb583c5e63e8ac130d6a377238052
Author: Eno Thereska 
Date:   2017-03-23T15:55:21Z

Merged with trunk

commit 75022449e2e4a1ac562504c63fe292c4d5586521
Author: Eno Thereska 
Date:   2017-03-23T16:02:15Z

Cleanup

commit 28297c56e5db02b7d34d6706d31d65e38f4b19b3
Author: Eno Thereska 
Date:   2017-03-23T16:03:40Z

Cleanup

commit 863e4e60b26ba7a718157863dc73a83779decb08
Author: Eno Thereska 
Date:   2017-03-23T16:40:49Z

Adjusted tests




---
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 #2725: KAFKA-4919: Document that stores must not be close...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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-4919) Document that stores must not be closed when Processors are closed

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Document that stores must not be closed when Processors are closed
> --
>
> Key: KAFKA-4919
> URL: https://issues.apache.org/jira/browse/KAFKA-4919
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Elias Levy
>Assignee: Damian Guy
>
> I have a streams job, that previously worked, that consumes and writes to a 
> large number of topics with many partitions and that uses many threads.  I 
> upgraded the job to 0.10.2.0.  The job now fails after a short time running, 
> seemingly after a rebalance.
> {quote}
> WARN  2017-03-19 18:03:20,432 [StreamThread-10][StreamThread.java:160] : 
> Unexpected state transition from RUNNING to NOT_RUNNING
> {quote}
> The first observation is that Streams is no longer outputting exceptions and 
> backtraces.  I had to add code to get this information.
> The exception:
> {quote}
> Exception: org.apache.kafka.streams.errors.StreamsException: Exception caught 
> in process. taskId=1_225, processor=KSTREAM-SOURCE-03, 
> topic=some_topic, partition=225, offset=266411
> org.apache.kafka.streams.errors.StreamsException: Exception caught in 
> process. taskId=1_225, processor=KSTREAM-SOURCE-03, topic=some_topic, 
> partition=225, offset=266411
>   at 
> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:641)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368)
> Caused by: org.apache.kafka.streams.errors.InvalidStateStoreException: Store 
> someStore-201701060400 is currently closed
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.validateStoreOpen(RocksDBStore.java:205)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.put(RocksDBStore.java:221)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.put(RocksDBSegmentedBytesStore.java:74)
>   at 
> org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.put(ChangeLoggingSegmentedBytesStore.java:54)
>   at 
> org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.put(MeteredSegmentedBytesStore.java:101)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:109)
>   ... X more
> {quote}
> The error occurs for many partitions.
> This was preceded by (for this partition):
> {quote}
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][ConsumerCoordinator.java:393] 
> : Revoking previously assigned partitions [some_topic-225] for group some_job
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:254] : 
> stream-thread [StreamThread-10] partitions [[some_topic-225]] revoked at the 
> beginning of consumer rebalance.
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:1056] : 
> stream-thread [StreamThread-10] Closing a task's topology 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:534] : 
> stream-thread [StreamThread-10] Committing consumer offsets of task 1_225
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1012] : 
> stream-thread [StreamThread-10] Updating suspended tasks to contain active 
> tasks [[1_225, 0_445, 0_30]]
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1019] : 
> stream-thread [StreamThread-10] Removing all active tasks [[1_225, 0_445, 
> 0_30]]
> INFO  2017-03-19 18:03:19,925 [StreamThread-10][ConsumerCoordinator.java:252] 
> : Setting newly assigned partitions [some_tpoic-225] for group some_job
> INFO  2017-03-19 18:03:19,927 [StreamThread-10][StreamThread.java:228] : 
> stream-thread [StreamThread-10] New partitions [[some_topic-225]] assigned at 
> the end of consumer rebalance.
> INFO  2017-03-19 18:03:19,929 [StreamThread-10][StreamTask.java:333] : task 
> [1_225] Initializing processor nodes of the topology
> Something happens.  What ???
> INFO  2017-03-19 18:03:20,135 [StreamThread-10][StreamThread.java:1045] : 
> stream-thread [StreamThread-10] Closing a task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:523] :

[jira] [Resolved] (KAFKA-4919) Document that stores must not be closed when Processors are closed

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-4919.
--
   Resolution: Fixed
Fix Version/s: 0.10.2.1
   0.11.0.0

> Document that stores must not be closed when Processors are closed
> --
>
> Key: KAFKA-4919
> URL: https://issues.apache.org/jira/browse/KAFKA-4919
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Elias Levy
>Assignee: Damian Guy
> Fix For: 0.11.0.0, 0.10.2.1
>
>
> I have a streams job, that previously worked, that consumes and writes to a 
> large number of topics with many partitions and that uses many threads.  I 
> upgraded the job to 0.10.2.0.  The job now fails after a short time running, 
> seemingly after a rebalance.
> {quote}
> WARN  2017-03-19 18:03:20,432 [StreamThread-10][StreamThread.java:160] : 
> Unexpected state transition from RUNNING to NOT_RUNNING
> {quote}
> The first observation is that Streams is no longer outputting exceptions and 
> backtraces.  I had to add code to get this information.
> The exception:
> {quote}
> Exception: org.apache.kafka.streams.errors.StreamsException: Exception caught 
> in process. taskId=1_225, processor=KSTREAM-SOURCE-03, 
> topic=some_topic, partition=225, offset=266411
> org.apache.kafka.streams.errors.StreamsException: Exception caught in 
> process. taskId=1_225, processor=KSTREAM-SOURCE-03, topic=some_topic, 
> partition=225, offset=266411
>   at 
> org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:641)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368)
> Caused by: org.apache.kafka.streams.errors.InvalidStateStoreException: Store 
> someStore-201701060400 is currently closed
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.validateStoreOpen(RocksDBStore.java:205)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStore.put(RocksDBStore.java:221)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBSegmentedBytesStore.put(RocksDBSegmentedBytesStore.java:74)
>   at 
> org.apache.kafka.streams.state.internals.ChangeLoggingSegmentedBytesStore.put(ChangeLoggingSegmentedBytesStore.java:54)
>   at 
> org.apache.kafka.streams.state.internals.MeteredSegmentedBytesStore.put(MeteredSegmentedBytesStore.java:101)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBWindowStore.put(RocksDBWindowStore.java:109)
>   ... X more
> {quote}
> The error occurs for many partitions.
> This was preceded by (for this partition):
> {quote}
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][ConsumerCoordinator.java:393] 
> : Revoking previously assigned partitions [some_topic-225] for group some_job
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:254] : 
> stream-thread [StreamThread-10] partitions [[some_topic-225]] revoked at the 
> beginning of consumer rebalance.
> INFO  2017-03-19 18:03:16,403 [StreamThread-10][StreamThread.java:1056] : 
> stream-thread [StreamThread-10] Closing a task's topology 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:17,887 [StreamThread-10][StreamThread.java:534] : 
> stream-thread [StreamThread-10] Committing consumer offsets of task 1_225
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1012] : 
> stream-thread [StreamThread-10] Updating suspended tasks to contain active 
> tasks [[1_225, 0_445, 0_30]]
> INFO  2017-03-19 18:03:17,891 [StreamThread-10][StreamThread.java:1019] : 
> stream-thread [StreamThread-10] Removing all active tasks [[1_225, 0_445, 
> 0_30]]
> INFO  2017-03-19 18:03:19,925 [StreamThread-10][ConsumerCoordinator.java:252] 
> : Setting newly assigned partitions [some_tpoic-225] for group some_job
> INFO  2017-03-19 18:03:19,927 [StreamThread-10][StreamThread.java:228] : 
> stream-thread [StreamThread-10] New partitions [[some_topic-225]] assigned at 
> the end of consumer rebalance.
> INFO  2017-03-19 18:03:19,929 [StreamThread-10][StreamTask.java:333] : task 
> [1_225] Initializing processor nodes of the topology
> Something happens.  What ???
> INFO  2017-03-19 18:03:20,135 [StreamThread-10][StreamThread.java:1045] : 
> stream-thread [StreamThread-10] Closing a task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:544] : 
> stream-thread [StreamThread-10] Flushing state stores of task 1_225
> INFO  2017-03-19 18:03:20,355 [StreamThread-10][StreamThread.java:523] : 
> stream-thread [StreamThr

[jira] [Commented] (KAFKA-4392) Failed to lock the state directory due to an unexpected exception

[Guozhang Wang (JIRA)]

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

Guozhang Wang commented on KAFKA-4392:
--

[~elevy] Could you share your settings (num.threads per process, num.partitions 
etc)? I suspect it was a different issue, e.g. the one we just got fixed in 
trunk: KAFKA-4848.

> Failed to lock the state directory due to an unexpected exception
> -
>
> Key: KAFKA-4392
> URL: https://issues.apache.org/jira/browse/KAFKA-4392
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.1.0
>Reporter: Ara Ebrahimi
>Assignee: Guozhang Wang
> Fix For: 0.10.2.0
>
>
> This happened on streaming startup, on a clean installation, no existing 
> folder. Here I was starting 4 instances of our streaming app on 4 machines 
> and one threw this exception. Seems to me there’s a race condition somewhere 
> when instances discover others, or something like that.
> 2016-11-02 15:43:47 INFO  StreamRunner:59 - Started http server successfully.
> 2016-11-02 15:44:50 ERROR StateDirectory:147 - Failed to lock the state 
> directory due to an unexpected exception
> java.nio.file.NoSuchFileException: 
> /data/1/kafka-streams/myapp-streams/7_21/.lock
>   at 
> sun.nio.fs.UnixException.translateToIOException(UnixException.java:86)
>   at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
>   at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
>   at 
> sun.nio.fs.UnixFileSystemProvider.newFileChannel(UnixFileSystemProvider.java:177)
>   at java.nio.channels.FileChannel.open(FileChannel.java:287)
>   at java.nio.channels.FileChannel.open(FileChannel.java:335)
>   at 
> org.apache.kafka.streams.processor.internals.StateDirectory.getOrCreateFileChannel(StateDirectory.java:176)
>   at 
> org.apache.kafka.streams.processor.internals.StateDirectory.lock(StateDirectory.java:90)
>   at 
> org.apache.kafka.streams.processor.internals.StateDirectory.cleanRemovedTasks(StateDirectory.java:140)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.maybeClean(StreamThread.java:552)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:459)
>   at 
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:242)
> ^C
> [arae@a4 ~]$ ls -al /data/1/kafka-streams/myapp-streams/7_21/
> ls: cannot access /data/1/kafka-streams/myapp-streams/7_21/: No such file or 
> directory
> [arae@a4 ~]$ ls -al /data/1/kafka-streams/myapp-streams/
> total 4
> drwxr-xr-x 74 root root 4096 Nov  2 15:44 .
> drwxr-xr-x  3 root root   27 Nov  2 15:43 ..
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_1
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_13
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_14
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_16
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_2
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_22
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_28
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_3
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_31
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_5
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_7
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_8
> drwxr-xr-x  3 root root   32 Nov  2 15:43 0_9
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_1
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_10
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_14
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_15
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_16
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_17
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_18
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_3
> drwxr-xr-x  3 root root   32 Nov  2 15:43 1_5
> drwxr-xr-x  3 root root   60 Nov  2 15:43 2_1
> drwxr-xr-x  3 root root   60 Nov  2 15:43 2_10
> drwxr-xr-x  3 root root   60 Nov  2 15:43 2_12
> drwxr-xr-x  3 root root   60 Nov  2 15:43 2_20
> drwxr-xr-x  3 root root   60 Nov  2 15:43 2_24
> drwxr-xr-x  3 root root   61 Nov  2 15:43 3_10
> drwxr-xr-x  3 root root   61 Nov  2 15:43 3_11
> drwxr-xr-x  3 root root   61 Nov  2 15:43 3_19
> drwxr-xr-x  3 root root   61 Nov  2 15:43 3_20
> drwxr-xr-x  3 root root   61 Nov  2 15:43 3_25
> drwxr-xr-x  3 root root   61 Nov  2 15:43 3_26
> drwxr-xr-x  3 root root   61 Nov  2 15:43 3_3
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_11
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_12
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_18
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_19
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_24
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_25
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_26
> drwxr-xr-x  3 root root   64 Nov  2 15:43 4_4
> drwxr-xr-x  3 root 

[GitHub] kafka pull request #2726: MINOR: only log first exception in RecordCollector...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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] KIP-134: Delay initial consumer group rebalance

[Guozhang Wang]

Just throwing another alternative idea here: we can consider using the
rebalance timeout value which is already included in the join request
protocol (and on the current Java client it is always written as the
session timeout value), that the first member joining will always force the
coordinator to wait that long. By doing this we do not need to bump up the
protocol either.


Guozhang

On Thu, Mar 23, 2017 at 5:49 AM, Damian Guy  wrote:

> Hi Ismael,
>
> Mostly to avoid the protocol bump.
>
> I agree that it may be difficult to choose the right delay for all consumer
> groups, but we wanted to make this something that most users don't really
> need to think about, i.e., a small enough default delay that works in the
> majority of cases. However it would be much more flexible as a consumer
> config, which i'm happy to pursue if this change is worthy of a protocol
> bump.
>
> Thanks,
> Damian
>
> On Thu, 23 Mar 2017 at 12:35 Ismael Juma  wrote:
>
> > Thanks for the KIP, Damian. It makes sense to avoid multiple rebalances
> > during start-up. One issue with having this as a broker config is that it
> > may be difficult to choose the right delay for all consumer groups. Can
> you
> > elaborate a little more on why the first alternative (add a consumer
> > config) was rejected? We bump protocol versions regularly (when it makes
> > sense), so it would be good to get a bit more detail.
> >
> > Thanks,
> > Ismael
> >
> > On Thu, Mar 23, 2017 at 12:24 PM, Damian Guy 
> wrote:
> >
> > > Hi All,
> > >
> > > I've prepared a KIP to add a configurable delay to the initial consumer
> > > group rebalance.
> > >
> > > Please have look here:
> > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > > 134%3A+Delay+initial+consumer+group+rebalance
> > >
> > > Thanks,
> > > Damian
> > >
> > > BTW, i apologize if this appears twice. Seems the first one may have
> not
> > > made it.
> > >
> >
>



-- 
-- Guozhang

Re: [DISCUSS] KIP-134: Delay initial consumer group rebalance

[Guozhang Wang]

The downside of it, though, is that although it "hides" this from most of
the users needing to be aware of it, by default session timeout i.e. the
rebalance timeout is 10 seconds which could arguably too long.


Guozhang

On Thu, Mar 23, 2017 at 10:12 AM, Guozhang Wang  wrote:

> Just throwing another alternative idea here: we can consider using the
> rebalance timeout value which is already included in the join request
> protocol (and on the current Java client it is always written as the
> session timeout value), that the first member joining will always force the
> coordinator to wait that long. By doing this we do not need to bump up the
> protocol either.
>
>
> Guozhang
>
> On Thu, Mar 23, 2017 at 5:49 AM, Damian Guy  wrote:
>
>> Hi Ismael,
>>
>> Mostly to avoid the protocol bump.
>>
>> I agree that it may be difficult to choose the right delay for all
>> consumer
>> groups, but we wanted to make this something that most users don't really
>> need to think about, i.e., a small enough default delay that works in the
>> majority of cases. However it would be much more flexible as a consumer
>> config, which i'm happy to pursue if this change is worthy of a protocol
>> bump.
>>
>> Thanks,
>> Damian
>>
>> On Thu, 23 Mar 2017 at 12:35 Ismael Juma  wrote:
>>
>> > Thanks for the KIP, Damian. It makes sense to avoid multiple rebalances
>> > during start-up. One issue with having this as a broker config is that
>> it
>> > may be difficult to choose the right delay for all consumer groups. Can
>> you
>> > elaborate a little more on why the first alternative (add a consumer
>> > config) was rejected? We bump protocol versions regularly (when it makes
>> > sense), so it would be good to get a bit more detail.
>> >
>> > Thanks,
>> > Ismael
>> >
>> > On Thu, Mar 23, 2017 at 12:24 PM, Damian Guy 
>> wrote:
>> >
>> > > Hi All,
>> > >
>> > > I've prepared a KIP to add a configurable delay to the initial
>> consumer
>> > > group rebalance.
>> > >
>> > > Please have look here:
>> > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
>> > > 134%3A+Delay+initial+consumer+group+rebalance
>> > >
>> > > Thanks,
>> > > Damian
>> > >
>> > > BTW, i apologize if this appears twice. Seems the first one may have
>> not
>> > > made it.
>> > >
>> >
>>
>
>
>
> --
> -- Guozhang
>



-- 
-- Guozhang

[jira] [Created] (KAFKA-4945) Suppress findbugs warnings about machine-generated code in jmh-benchmarks

[Colin P. McCabe (JIRA)]

Colin P. McCabe created KAFKA-4945:
--

 Summary: Suppress findbugs warnings about machine-generated code 
in jmh-benchmarks
 Key: KAFKA-4945
 URL: https://issues.apache.org/jira/browse/KAFKA-4945
 Project: Kafka
  Issue Type: Bug
Reporter: Colin P. McCabe
Assignee: Colin P. McCabe


Suppress findbugs warnings about some machine-generated code in jmh-benchmarks. 
 The warnings are about variables that are set, but not later read.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [DISCUSS] KIP-134: Delay initial consumer group rebalance

[Damian Guy]

Thanks Gouzhang - i think another problem with this is that is overloading
session.timeout.ms to mean multiple things. I'm not sure that is a good
thing.

On Thu, 23 Mar 2017 at 17:14 Guozhang Wang  wrote:

> The downside of it, though, is that although it "hides" this from most of
> the users needing to be aware of it, by default session timeout i.e. the
> rebalance timeout is 10 seconds which could arguably too long.
>
>
> Guozhang
>
> On Thu, Mar 23, 2017 at 10:12 AM, Guozhang Wang 
> wrote:
>
> > Just throwing another alternative idea here: we can consider using the
> > rebalance timeout value which is already included in the join request
> > protocol (and on the current Java client it is always written as the
> > session timeout value), that the first member joining will always force
> the
> > coordinator to wait that long. By doing this we do not need to bump up
> the
> > protocol either.
> >
> >
> > Guozhang
> >
> > On Thu, Mar 23, 2017 at 5:49 AM, Damian Guy 
> wrote:
> >
> >> Hi Ismael,
> >>
> >> Mostly to avoid the protocol bump.
> >>
> >> I agree that it may be difficult to choose the right delay for all
> >> consumer
> >> groups, but we wanted to make this something that most users don't
> really
> >> need to think about, i.e., a small enough default delay that works in
> the
> >> majority of cases. However it would be much more flexible as a consumer
> >> config, which i'm happy to pursue if this change is worthy of a protocol
> >> bump.
> >>
> >> Thanks,
> >> Damian
> >>
> >> On Thu, 23 Mar 2017 at 12:35 Ismael Juma  wrote:
> >>
> >> > Thanks for the KIP, Damian. It makes sense to avoid multiple
> rebalances
> >> > during start-up. One issue with having this as a broker config is that
> >> it
> >> > may be difficult to choose the right delay for all consumer groups.
> Can
> >> you
> >> > elaborate a little more on why the first alternative (add a consumer
> >> > config) was rejected? We bump protocol versions regularly (when it
> makes
> >> > sense), so it would be good to get a bit more detail.
> >> >
> >> > Thanks,
> >> > Ismael
> >> >
> >> > On Thu, Mar 23, 2017 at 12:24 PM, Damian Guy 
> >> wrote:
> >> >
> >> > > Hi All,
> >> > >
> >> > > I've prepared a KIP to add a configurable delay to the initial
> >> consumer
> >> > > group rebalance.
> >> > >
> >> > > Please have look here:
> >> > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> >> > > 134%3A+Delay+initial+consumer+group+rebalance
> >> > >
> >> > > Thanks,
> >> > > Damian
> >> > >
> >> > > BTW, i apologize if this appears twice. Seems the first one may have
> >> not
> >> > > made it.
> >> > >
> >> >
> >>
> >
> >
> >
> > --
> > -- Guozhang
> >
>
>
>
> --
> -- Guozhang
>

[GitHub] kafka pull request #2729: KAFKA-4945: Suppress findbugs warnings about machi...

[cmccabe]

GitHub user cmccabe opened a pull request:

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

KAFKA-4945: Suppress findbugs warnings about machine-generated code i…

…n jmh-benchmarks

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

$ git pull https://github.com/cmccabe/kafka KAFKA-4945

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

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


commit dc4146324b49c8fd5068e15723c37e87debecb7b
Author: Colin P. Mccabe 
Date:   2017-03-23T17:15:21Z

KAFKA-4945: Suppress findbugs warnings about machine-generated code in 
jmh-benchmarks




---
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-4945) Suppress findbugs warnings about machine-generated code in jmh-benchmarks

[ASF GitHub Bot (JIRA)]

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

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

GitHub user cmccabe opened a pull request:

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

KAFKA-4945: Suppress findbugs warnings about machine-generated code i…

…n jmh-benchmarks

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

$ git pull https://github.com/cmccabe/kafka KAFKA-4945

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

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


commit dc4146324b49c8fd5068e15723c37e87debecb7b
Author: Colin P. Mccabe 
Date:   2017-03-23T17:15:21Z

KAFKA-4945: Suppress findbugs warnings about machine-generated code in 
jmh-benchmarks




> Suppress findbugs warnings about machine-generated code in jmh-benchmarks
> -
>
> Key: KAFKA-4945
> URL: https://issues.apache.org/jira/browse/KAFKA-4945
> Project: Kafka
>  Issue Type: Bug
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
>
> Suppress findbugs warnings about some machine-generated code in 
> jmh-benchmarks.  The warnings are about variables that are set, but not later 
> read.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4938) Creating a connector with missing name parameter throws a NullPointerException

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-4938:
-
Labels: newbie  (was: )

> Creating a connector with missing name parameter throws a NullPointerException
> --
>
> Key: KAFKA-4938
> URL: https://issues.apache.org/jira/browse/KAFKA-4938
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Sönke Liebau
>Priority: Minor
>  Labels: newbie
>
> Creating a connector via the rest api runs into a NullPointerException, when 
> omitting the name parameter in the request.
> {code}
> POST 127.0.0.1:8083/connectors
> {
>   "config": {
> "connector.class": "org.apache.kafka.connect.tools.MockSourceConnector",
> "tasks.max": "1",
> "topics": "test-topic"
>   }
> }
> {code}
> Results in a 500 return code, due to a NullPointerException being thrown when 
> checking the name for slashes 
> [here|https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java#L91].
>  I believe this was introduced with the fix for 
> [KAFKA-4372|https://issues.apache.org/jira/browse/KAFKA-4372]



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4278) Undocumented REST resources

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-4278:
-
Labels: newbie  (was: )

> Undocumented REST resources
> ---
>
> Key: KAFKA-4278
> URL: https://issues.apache.org/jira/browse/KAFKA-4278
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Gwen Shapira
>  Labels: newbie
>
> We've added some REST resources and I think we didn't document them.
> / - get version
> /connector-plugins - show installed connectors
> Those are the ones I've found (or rather, failed to find) - there could be 
> more.
> Perhaps the best solution is to auto-generate the REST documentation the way 
> we generate configuration docs?



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4942) Kafka Connect: Offset committing times out before expected

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava commented on KAFKA-4942:
--

[~stephane.maa...@gmail.com] Any more context from earlier in the logs? 
Definitely looks fishy, and I don't see a way that could happen where the log 
message would just be misleading (it happens here 
https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L171
 after a condition, so it's not anything like an unrelated exception).

> Kafka Connect: Offset committing times out before expected
> --
>
> Key: KAFKA-4942
> URL: https://issues.apache.org/jira/browse/KAFKA-4942
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Stephane Maarek
>
> On Kafka 0.10.2.0
> I run a connector that deals with a lot of data, in a kafka connect cluster
> When the offsets are getting committed, I get the following:
> {code}
> [2017-03-23 03:56:25,134] INFO WorkerSinkTask{id=MyConnector-1} Committing 
> offsets (org.apache.kafka.connect.runtime.WorkerSinkTask)
> [2017-03-23 03:56:25,135] WARN Commit of WorkerSinkTask{id=MyConnector-1} 
> offsets timed out (org.apache.kafka.connect.runtime.WorkerSinkTask)
> {code}
> If you look at the timestamps, they're 1 ms apart. My settings are the 
> following: 
> {code}
>   offset.flush.interval.ms = 12
>   offset.flush.timeout.ms = 6
>   offset.storage.topic = _connect_offsets
> {code}
> It seems the offset flush timeout setting is completely ignored for the look 
> of the logs. I would expect the timeout message to happen 60 seconds after 
> the commit offset INFO message, not 1 millisecond later.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4400) Prefix for sink task consumer groups should be configurable

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-4400:
-
Labels: newbie  (was: )

> Prefix for sink task consumer groups should be configurable
> ---
>
> Key: KAFKA-4400
> URL: https://issues.apache.org/jira/browse/KAFKA-4400
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.1.0
>Reporter: Ewen Cheslack-Postava
>  Labels: newbie
>
> Currently the prefix for creating consumer groups is fixed. This means that 
> if you run multiple Connect clusters using the same Kafka cluster and create 
> connectors with the same name, sink tasks in different clusters will join the 
> same group. Making this prefix configurable at the worker level would protect 
> against this.
> An alternative would be to define unique cluster IDs for each connect 
> cluster, which would allow us to construct a unique name for the group 
> without requiring yet another config (but presents something of a 
> compatibility challenge).



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4855) Struct SchemaBuilder should not allow duplicate fields.

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-4855:
-
Labels: newbie  (was: )

> Struct SchemaBuilder should not allow duplicate fields.
> ---
>
> Key: KAFKA-4855
> URL: https://issues.apache.org/jira/browse/KAFKA-4855
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Jeremy Custenborder
>  Labels: newbie
>
> I would expect this to fail at the build() on schema. It actually makes it 
> all the way to Struct.validate() and throws a cryptic error message. .field() 
> should throw an exception if a field is already used.
> Repro:
> {code}
>   @Test
>   public void duplicateFields() {
> final Schema schema = SchemaBuilder.struct()
> .name("testing")
> .field("id", SchemaBuilder.string().doc("").build())
> .field("id", SchemaBuilder.string().doc("").build())
> .build();
> final Struct struct = new Struct(schema)
> .put("id", "testing");
> struct.validate();
>   }
> {code}
> {code}
> org.apache.kafka.connect.errors.DataException: Invalid value: null used for 
> required field at 
> org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:212)
>   at org.apache.kafka.connect.data.Struct.validate(Struct.java:232)
>   at 
> io.confluent.kafka.connect.jms.RecordConverterTest.duplicateFieldRepro(RecordConverterTest.java:289)
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4280) Document REST resource for showing available connector plugin configs

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-4280:
-
Summary: Document REST resource for showing available connector plugin 
configs  (was: Add REST resource for showing available connector plugin configs)

> Document REST resource for showing available connector plugin configs
> -
>
> Key: KAFKA-4280
> URL: https://issues.apache.org/jira/browse/KAFKA-4280
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Gwen Shapira
>  Labels: newbie
>
> Connector-plugins allow listing the plugs and validating configs, but we have 
> nothing (I think?) for listing available configuration properties.
> If this doesn't exist, would be good for usability to add it. If it does 
> exist, perhaps document it?



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4810) SchemaBuilder should be more lax about checking that fields are unset if they are being set to the same value

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava updated KAFKA-4810:
-
Labels: newbie  (was: )

> SchemaBuilder should be more lax about checking that fields are unset if they 
> are being set to the same value
> -
>
> Key: KAFKA-4810
> URL: https://issues.apache.org/jira/browse/KAFKA-4810
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Ewen Cheslack-Postava
>  Labels: newbie
>   Original Estimate: 1h
>  Remaining Estimate: 1h
>
> Currently SchemaBuilder is strict when checking that certain fields have not 
> been set yet (e.g. version, name, doc). It just checks that the field is 
> null. This is intended to protect the user from buggy code that overwrites a 
> field with different values, but it's a bit too strict currently. In generic 
> code for converting schemas (e.g. Converters) you will sometimes initialize a 
> builder with these values (e.g. because you get a SchemaBuilder for a logical 
> type, which sets name & version), but then have generic code for setting name 
> & version from the source schema.
> We saw this bug in practice with Confluent's AvroConverter, so it's likely it 
> could trip up others as well. You can work around the issue, but it would be 
> nice if exceptions were only thrown if you try to overwrite an existing value 
> with a different value.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [VOTE] KIP-122: Add Reset Consumer Group Offsets tooling

[Jorge Esteban Quilcate Otoya]

@Ismael, thanks for your feedback!
1. Good point. I will add optional support for timezone as part of the
datetime input. But, when datetime is without timezone, would it be more
consistent to get the timezone from the cluster first and then reset based
on that value? Not sure if it is possible to get that info from the
cluster. But, in case that's not available, I could add a note to advise
that in case timezone is not specified the tool will get that value from
the client and it would be user's responsibility to validate that is
aligned with the server.
2. Happy to add it to the KIP.
3. This was part of the discussion thread, we end up with `shift-by` to
avoid adding `offset` to each case and make it a bit more consistent.
4. I could remove CURRENT-OFFSET and CURRENT-LAG from the output, and leave
it as part of `describe` operation, if that's better.
5. Agree. At the beginning we consider `shift-plus` and `shift-minus`, but
agree to join them in one option and accept +/- as input. Maybe that's a
better option?

El jue., 23 mar. 2017 a las 17:17, Ismael Juma ()
escribió:

Hi Jorge,

Thanks for the detailed KIP. The tool looks very useful. A few comments:

1. We are using the default timezone of the client for the specified date.
This seems a bit error prone. Would it be better to require the users to
specify the time zone as part of the date time? We should at least allow
it, but my experience when it comes to using the default time zone in a
distributed environment is not great.
2. It seems like we are using the ISO 8601 format for date time and
duration. It would be good to mention that.
3. `shift-by` should perhaps be `shift-offset-by` to be a bit clearer.
4. Printing the current offset via reset-offsets is a bit odd, can we not
use a different option for that?
5. It's a bit odd that `by-duration` subtracts while `shift-by` moves
forward. It would be nice if the name made it clear that `by-duration` is
subtracting, but I have no good suggestions, so maybe that's the best we
can do.

Ismael

On Fri, Feb 24, 2017 at 12:46 AM, Jorge Esteban Quilcate Otoya <
quilcate.jo...@gmail.com> wrote:

> Hi All,
>
> It seems that there is no further concern with the KIP-122.
> At this point we would like to start the voting process.
>
> The KIP can be found here:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 122%3A+Add+Reset+Consumer+Group+Offsets+tooling
>
>
> Thanks!
>
> Jorge.
>

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

[Apache Jenkins Server]

See 

Jenkins build is back to normal : kafka-0.10.2-jdk7 #112

[Apache Jenkins Server]

See 

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

[Apache Jenkins Server]

See 


Changes:

[wangguoz] MINOR: only log first exception in RecordCollectorImpl producer 
callback

--
[...truncated 165.14 KB...]

kafka.security.auth.SimpleAclAuthorizerTest > testLoadCache STARTED

kafka.security.auth.SimpleAclAuthorizerTest > testLoadCache PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testIsrAfterBrokerShutDownAndJoinsBack STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testIsrAfterBrokerShutDownAndJoinsBack PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAutoCreateTopicWithCollision STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAutoCreateTopicWithCollision PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokerListWithNoTopics STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokerListWithNoTopics PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > testAutoCreateTopic STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > testAutoCreateTopic PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > testGetAllTopicMetadata 
STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > testGetAllTopicMetadata 
PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > testBasicTopicMetadata 
STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > testBasicTopicMetadata PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAutoCreateTopicWithInvalidReplication STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAutoCreateTopicWithInvalidReplication PASSED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown STARTED

kafka.integration.SaslPlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown PASSED

kafka.integration.PrimitiveApiTest > testMultiProduce STARTED

kafka.integration.PrimitiveApiTest > testMultiProduce PASSED

kafka.integration.PrimitiveApiTest > testDefaultEncoderProducerAndFetch STARTED

kafka.integration.PrimitiveApiTest > testDefaultEncoderProducerAndFetch PASSED

kafka.integration.PrimitiveApiTest > testFetchRequestCanProperlySerialize 
STARTED

kafka.integration.PrimitiveApiTest > testFetchRequestCanProperlySerialize PASSED

kafka.integration.PrimitiveApiTest > testPipelinedProduceRequests STARTED

kafka.integration.PrimitiveApiTest > testPipelinedProduceRequests PASSED

kafka.integration.PrimitiveApiTest > testProduceAndMultiFetch STARTED

kafka.integration.PrimitiveApiTest > testProduceAndMultiFetch PASSED

kafka.integration.PrimitiveApiTest > 
testDefaultEncoderProducerAndFetchWithCompression STARTED

kafka.integration.PrimitiveApiTest > 
testDefaultEncoderProducerAndFetchWithCompression PASSED

kafka.integration.PrimitiveApiTest > testConsumerEmptyTopic STARTED

kafka.integration.PrimitiveApiTest > testConsumerEmptyTopic PASSED

kafka.integration.PrimitiveApiTest > testEmptyFetchRequest STARTED

kafka.integration.PrimitiveApiTest > testEmptyFetchRequest PASSED

kafka.integration.PlaintextTopicMetadataTest > 
testIsrAfterBrokerShutDownAndJoinsBack STARTED

kafka.integration.PlaintextTopicMetadataTest > 
testIsrAfterBrokerShutDownAndJoinsBack PASSED

kafka.integration.PlaintextTopicMetadataTest > testAutoCreateTopicWithCollision 
STARTED

kafka.integration.PlaintextTopicMetadataTest > testAutoCreateTopicWithCollision 
PASSED

kafka.integration.PlaintextTopicMetadataTest > testAliveBrokerListWithNoTopics 
STARTED

kafka.integration.PlaintextTopicMetadataTest > testAliveBrokerListWithNoTopics 
PASSED

kafka.integration.PlaintextTopicMetadataTest > testAutoCreateTopic STARTED

kafka.integration.PlaintextTopicMetadataTest > testAutoCreateTopic PASSED

kafka.integration.PlaintextTopicMetadataTest > testGetAllTopicMetadata STARTED

kafka.integration.PlaintextTopicMetadataTest > testGetAllTopicMetadata PASSED

kafka.integration.PlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup STARTED

kafka.integration.PlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup PASSED

kafka.integration.PlaintextTopicMetadataTest > testBasicTopicMetadata STARTED

kafka.integration.PlaintextTopicMetadataTest > testBasicTopicMetadata PASSED

kafka.integration.PlaintextTopicMetadataTest > 
testAutoCreateTopicWithInvalidReplication STARTED

kafka.integration.PlaintextTopicMetadataTest > 
testAutoCreateTopicWithInvalidReplication PASSED

kafka.integration.PlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown STARTED

kafka.integration.PlaintextTopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown PASSED

kafka.integration.Me

[jira] [Work started] (KAFKA-4916) Add streams tests with brokers failing

[Eno Thereska (JIRA)]

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

Work on KAFKA-4916 started by Eno Thereska.
---
> Add streams tests with brokers failing
> --
>
> Key: KAFKA-4916
> URL: https://issues.apache.org/jira/browse/KAFKA-4916
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
>Assignee: Eno Thereska
>Priority: Critical
> Fix For: 0.11.0.0
>
>
> We need to add either integration or system tests with streams and have Kafka 
> brokers fail and come back up. A combination of transient and permanent 
> broker failures.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [DISCUSS] KIP-120: Cleanup Kafka Streams builder API

[Matthias J. Sax]

Jay,

about the naming schema:

>>1. "kstreams" - the DSL
>>2. "processor api" - the lower level callback/topology api
>>3. KStream/KTable - entities in the kstreams dsl
>>4. "Kafka Streams" - General name for stream processing stuff in Kafka,
>>including both kstreams and the processor API plus the underlying
>>implementation.

It think this terminology has some issues... To me, `kstreams` was
always not more than an abbreviation for `Kafka Streams` -- thus (1) and
(4) kinda collide here. Following questions on the mailing list etc I
often see people using kstreams or kstream exactly a abbr. for "Kafka
Streams"

> I think referring to the dsl as "kstreams" is cute and pneumonic and not
> particularly confusing.

I disagree here. It's a very subtle difference between `kstreams` and
`KStream` -- just singular/plural, thus (1) and (3) also "collide" --
it's just too close to each other.

Thus, I really think it's a good idea to get a new name for the DSL to
get a better separation of the 4 concepts.

Furthermore, we use the term "Streams API". Thus, I think
`StreamsBuilder` (or `StreamsTopologyBuilder`) are both very good names.


Thus, I prefer to keep the KIP as is (suggesting `StreamsBuilder`).

I will start a VOTE thread. Of course, we can still discuss the naming
issue. :)



-Matthias


On 3/22/17 8:53 PM, Jay Kreps wrote:
> I don't feel strongly on this, so I'm happy with whatever everyone else
> wants.
> 
> Michael, I'm not arguing that people don't need to understand topologies, I
> just think it is like rocks db, you need to understand it when
> debugging/operating but not in the initial coding since the metaphor we're
> providing at this layer isn't a topology of processors but rather something
> like the collections api. Anyhow it won't hurt people to have it there.
> 
> For the original KStreamBuilder thing, I think that came from the naming we
> discussed originally:
> 
>1. "kstreams" - the DSL
>2. "processor api" - the lower level callback/topology api
>3. KStream/KTable - entities in the kstreams dsl
>4. "Kafka Streams" - General name for stream processing stuff in Kafka,
>including both kstreams and the processor API plus the underlying
>implementation.
> 
> I think referring to the dsl as "kstreams" is cute and pneumonic and not
> particularly confusing. Just like referring to the "java collections
> library" isn't confusing even though it contains the Iterator interface
> which is not actually itself a collection.
> 
> So I think KStreamBuilder should technically have been KstreamsBuilder and
> is intended not to be a builder of a KStream but rather the builder for the
> kstreams DSL. Okay, yes, that *is* slightly confusing. :-)
> 
> -Jay
> 
> On Wed, Mar 22, 2017 at 11:25 AM, Guozhang Wang  wrote:
> 
>> Regarding the naming of `StreamsTopologyBuilder` v.s. `StreamsBuilder` that
>> are going to be used in DSL, I agree both has their arguments:
>>
>> 1. On one side, people using the DSL layer probably do not need to be aware
>> (or rather, "learn about") of the "topology" concept, although this concept
>> is a publicly exposed one in Kafka Streams.
>>
>> 2. On the other side, StreamsBuilder#build() returning a Topology object
>> sounds a little weird, at least to me (admittedly subjective matter).
>>
>>
>> Since the second bullet point seems to be more "subjective" and many people
>> are not worried about it, I'm OK to go with the other option.
>>
>>
>> Guozhang
>>
>>
>> On Wed, Mar 22, 2017 at 8:58 AM, Michael Noll 
>> wrote:
>>
>>> Forwarding to kafka-user.
>>>
>>>
>>> -- Forwarded message --
>>> From: Michael Noll 
>>> Date: Wed, Mar 22, 2017 at 8:48 AM
>>> Subject: Re: [DISCUSS] KIP-120: Cleanup Kafka Streams builder API
>>> To: dev@kafka.apache.org
>>>
>>>
>>> Matthias,
>>>
 @Michael:

 You seemed to agree with Jay about not exposing the `Topology` concept
 in our main entry class (ie, current KStreamBuilder), thus, I
 interpreted that you do not want `Topology` in the name either (I am a
 little surprised by your last response, that goes the opposite
>>> direction).
>>>
>>> Oh, sorry for not being clear.
>>>
>>> What I wanted to say in my earlier email was the following:  Yes, I do
>>> agree with most of Jay's reasoning, notably about carefully deciding how
>>> much and which parts of the API/concept "surface" we expose to users of
>> the
>>> DSL.  However, and this is perhaps where I wasn't very clear, I disagree
>> on
>>> the particular opinion about not exposing the topology concept to DSL
>>> users.  Instead, I think the concept of a topology is important to
>>> understand even for DSL users -- particularly because of the way the DSL
>> is
>>> currently wiring your processing logic via the builder pattern.  (As I
>>> noted, e.g. Akka uses a different approach where you might be able to get
>>> away with not exposing the "topology" concept, but even in Akka there's
>> the
>>> notion of graphs an

[VOTE] KIP-120: Cleanup Kafka Streams builder API

[Matthias J. Sax]

Hi,

I would like to start the VOTE on KIP-120:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-120%3A+Cleanup+Kafka+Streams+builder+API

If you have further comments, please reply to the DISCUSS thread.

Thanks a lot!


-Matthias



signature.asc
Description: OpenPGP digital signature

Build failed in Jenkins: kafka-0.10.2-jdk7 #113

[Apache Jenkins Server]

See 


Changes:

[wangguoz] MINOR: only log first exception in RecordCollectorImpl producer 
callback

--
[...truncated 754.59 KB...]
org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowExceptionOverlappingTopic STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowExceptionOverlappingTopic PASSED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowStreamsExceptionNoResetSpecified STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowStreamsExceptionNoResetSpecified PASSED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldOnlyReadRecordsWhereEarliestSpecified STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldOnlyReadRecordsWhereEarliestSpecified PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldBeAbleToQueryState[0] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldBeAbleToQueryState[0] PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldNotMakeStoreAvailableUntilAllStoresAvailable[0] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldNotMakeStoreAvailableUntilAllStoresAvailable[0] PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
queryOnRebalance[0] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
queryOnRebalance[0] PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
concurrentAccesses[0] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
concurrentAccesses[0] PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldBeAbleToQueryState[1] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldBeAbleToQueryState[1] PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldNotMakeStoreAvailableUntilAllStoresAvailable[1] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
shouldNotMakeStoreAvailableUntilAllStoresAvailable[1] PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
queryOnRebalance[1] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
queryOnRebalance[1] PASSED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
concurrentAccesses[1] STARTED

org.apache.kafka.streams.integration.QueryableStateIntegrationTest > 
concurrentAccesses[1] PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testInnerKTableKTable STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testInnerKTableKTable PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > testLeftKTableKTable 
STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > testLeftKTableKTable 
PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testLeftKStreamKStream STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testLeftKStreamKStream PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testLeftKStreamKTable STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testLeftKStreamKTable PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testOuterKTableKTable STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testOuterKTableKTable PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testInnerKStreamKStream STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testInnerKStreamKStream PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testOuterKStreamKStream STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testOuterKStreamKStream PASSED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testInnerKStreamKTable STARTED

org.apache.kafka.streams.integration.JoinIntegrationTest > 
testInnerKStreamKTable PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableJoin PASSED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
testReprocessingFromScratchAfterResetWithIntermediateUserTopic STARTED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
testReprocessingFromScratchAfterResetWithIntermediateUserTopic PASSED

org.apache.kafka.streams.int

Re: [DISCUSS] KIP-134: Delay initial consumer group rebalance

[Matthias J. Sax]

Thanks for the KIP Damian!

My two cents:

 - we should have an explicit parameter for this -- implicit setting are
always tricky (the "importance" of this parameter would be LOW)

 - the config should be different for each consumer group:
   * assume you have a stateless app, you want to rebalance immediately
   * if you start-up in an visualized environment using some tools like
Mesos you might need a different value that on bare metal (no VM to be
started)
   * it also depends, how many consumer instanced you expect -- it's
harder to start up 100 instances in 3 seconds than 5

 - the default value should be zero


One more thought: what about scaling scenarios? If a consumer group has
10 instanced and should be scaled up to 20, it would make sense to do
this with a single rebalance, too. Thus, I am wondering, if it would
make sense to apply this delay each time a new consumer joins group,
even if the group is not empty?


-Matthias


On 3/23/17 10:19 AM, Damian Guy wrote:
> Thanks Gouzhang - i think another problem with this is that is overloading
> session.timeout.ms to mean multiple things. I'm not sure that is a good
> thing.
> 
> On Thu, 23 Mar 2017 at 17:14 Guozhang Wang  wrote:
> 
>> The downside of it, though, is that although it "hides" this from most of
>> the users needing to be aware of it, by default session timeout i.e. the
>> rebalance timeout is 10 seconds which could arguably too long.
>>
>>
>> Guozhang
>>
>> On Thu, Mar 23, 2017 at 10:12 AM, Guozhang Wang 
>> wrote:
>>
>>> Just throwing another alternative idea here: we can consider using the
>>> rebalance timeout value which is already included in the join request
>>> protocol (and on the current Java client it is always written as the
>>> session timeout value), that the first member joining will always force
>> the
>>> coordinator to wait that long. By doing this we do not need to bump up
>> the
>>> protocol either.
>>>
>>>
>>> Guozhang
>>>
>>> On Thu, Mar 23, 2017 at 5:49 AM, Damian Guy 
>> wrote:
>>>
 Hi Ismael,

 Mostly to avoid the protocol bump.

 I agree that it may be difficult to choose the right delay for all
 consumer
 groups, but we wanted to make this something that most users don't
>> really
 need to think about, i.e., a small enough default delay that works in
>> the
 majority of cases. However it would be much more flexible as a consumer
 config, which i'm happy to pursue if this change is worthy of a protocol
 bump.

 Thanks,
 Damian

 On Thu, 23 Mar 2017 at 12:35 Ismael Juma  wrote:

> Thanks for the KIP, Damian. It makes sense to avoid multiple
>> rebalances
> during start-up. One issue with having this as a broker config is that
 it
> may be difficult to choose the right delay for all consumer groups.
>> Can
 you
> elaborate a little more on why the first alternative (add a consumer
> config) was rejected? We bump protocol versions regularly (when it
>> makes
> sense), so it would be good to get a bit more detail.
>
> Thanks,
> Ismael
>
> On Thu, Mar 23, 2017 at 12:24 PM, Damian Guy 
 wrote:
>
>> Hi All,
>>
>> I've prepared a KIP to add a configurable delay to the initial
 consumer
>> group rebalance.
>>
>> Please have look here:
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
>> 134%3A+Delay+initial+consumer+group+rebalance
>>
>> Thanks,
>> Damian
>>
>> BTW, i apologize if this appears twice. Seems the first one may have
 not
>> made it.
>>
>

>>>
>>>
>>>
>>> --
>>> -- Guozhang
>>>
>>
>>
>>
>> --
>> -- Guozhang
>>
> 



signature.asc
Description: OpenPGP digital signature

[jira] [Commented] (KAFKA-4942) Kafka Connect: Offset committing times out before expected

[Stephane Maarek (JIRA)]

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

Stephane Maarek commented on KAFKA-4942:


[~ewencp] nothing weird from the log, except that EVERY offset commit gives me 
that behaviour. You know what's even weirder? Offsets still are committed. It's 
because the line you linked me doesn't interrupt the offset commit, it just 
throws a warning.
Now the reason I would see that line being triggered right away is when 
now >= commitTimeoutMs
now >= commitStarted + 
workerConfig.getLong(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG)

I believe the setting workerConfig.getLong should return something 6 ms, so 
it hints towards commitStarted being corrupted. 

And I think I just found your issue.
commitTimeoutMs happens before the try block.
so no matter what happens in the commitOffsets(now, false); , it won't impact 
the comparison:
 if (committing && now >= commitTimeoutMs) 

And that's the bug :)

> Kafka Connect: Offset committing times out before expected
> --
>
> Key: KAFKA-4942
> URL: https://issues.apache.org/jira/browse/KAFKA-4942
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Stephane Maarek
>
> On Kafka 0.10.2.0
> I run a connector that deals with a lot of data, in a kafka connect cluster
> When the offsets are getting committed, I get the following:
> {code}
> [2017-03-23 03:56:25,134] INFO WorkerSinkTask{id=MyConnector-1} Committing 
> offsets (org.apache.kafka.connect.runtime.WorkerSinkTask)
> [2017-03-23 03:56:25,135] WARN Commit of WorkerSinkTask{id=MyConnector-1} 
> offsets timed out (org.apache.kafka.connect.runtime.WorkerSinkTask)
> {code}
> If you look at the timestamps, they're 1 ms apart. My settings are the 
> following: 
> {code}
>   offset.flush.interval.ms = 12
>   offset.flush.timeout.ms = 6
>   offset.storage.topic = _connect_offsets
> {code}
> It seems the offset flush timeout setting is completely ignored for the look 
> of the logs. I would expect the timeout message to happen 60 seconds after 
> the commit offset INFO message, not 1 millisecond later.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Comment Edited] (KAFKA-4942) Kafka Connect: Offset committing times out before expected

[Stephane Maarek (JIRA)]

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

Stephane Maarek edited comment on KAFKA-4942 at 3/23/17 11:57 PM:
--

[~ewencp] nothing weird from the log, except that EVERY offset commit gives me 
that behaviour. You know what's even weirder? Offsets still are committed. It's 
because the line you linked me doesn't interrupt the offset commit, it just 
throws a warning.
Now the reason I would see that line being triggered right away is when 
now >= commitTimeoutMs
now >= commitStarted + 
workerConfig.getLong(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG)

I believe the setting workerConfig.getLong should return something 6 ms, so 
it hints towards commitStarted being corrupted. 

And I think I just found your issue.
commitTimeoutMs happens before the try block.
so no matter what happens in the commitOffsets(now, false); , it won't impact 
the comparison:
 if (committing && now >= commitTimeoutMs) 

And that's the bug :) One offset that times out on your thread and all the 
subsequent tries will show that warning 


was (Author: stephane.maa...@gmail.com):
[~ewencp] nothing weird from the log, except that EVERY offset commit gives me 
that behaviour. You know what's even weirder? Offsets still are committed. It's 
because the line you linked me doesn't interrupt the offset commit, it just 
throws a warning.
Now the reason I would see that line being triggered right away is when 
now >= commitTimeoutMs
now >= commitStarted + 
workerConfig.getLong(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG)

I believe the setting workerConfig.getLong should return something 6 ms, so 
it hints towards commitStarted being corrupted. 

And I think I just found your issue.
commitTimeoutMs happens before the try block.
so no matter what happens in the commitOffsets(now, false); , it won't impact 
the comparison:
 if (committing && now >= commitTimeoutMs) 

And that's the bug :)

> Kafka Connect: Offset committing times out before expected
> --
>
> Key: KAFKA-4942
> URL: https://issues.apache.org/jira/browse/KAFKA-4942
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Stephane Maarek
>
> On Kafka 0.10.2.0
> I run a connector that deals with a lot of data, in a kafka connect cluster
> When the offsets are getting committed, I get the following:
> {code}
> [2017-03-23 03:56:25,134] INFO WorkerSinkTask{id=MyConnector-1} Committing 
> offsets (org.apache.kafka.connect.runtime.WorkerSinkTask)
> [2017-03-23 03:56:25,135] WARN Commit of WorkerSinkTask{id=MyConnector-1} 
> offsets timed out (org.apache.kafka.connect.runtime.WorkerSinkTask)
> {code}
> If you look at the timestamps, they're 1 ms apart. My settings are the 
> following: 
> {code}
>   offset.flush.interval.ms = 12
>   offset.flush.timeout.ms = 6
>   offset.storage.topic = _connect_offsets
> {code}
> It seems the offset flush timeout setting is completely ignored for the look 
> of the logs. I would expect the timeout message to happen 60 seconds after 
> the commit offset INFO message, not 1 millisecond later.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2730: KAFKA-4942 fixed the commitTimeoutMs being set bef...

[simplesteph]

GitHub user simplesteph opened a pull request:

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

KAFKA-4942 fixed the commitTimeoutMs being set before the commit actually 
started

this fixes KAFKA-4942

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

$ git pull https://github.com/simplesteph/kafka fix-connect-offset-commit

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

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


commit 4d1db91ec8276da3dcca6d85b66c70497cf0815d
Author: simplesteph 
Date:   2017-03-24T00:03:07Z

fixed the commitTimeoutMs being set before the commit actually started
this fixes KAFKA-4942




---
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-4942) Kafka Connect: Offset committing times out before expected

[ASF GitHub Bot (JIRA)]

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

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

GitHub user simplesteph opened a pull request:

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

KAFKA-4942 fixed the commitTimeoutMs being set before the commit actually 
started

this fixes KAFKA-4942

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

$ git pull https://github.com/simplesteph/kafka fix-connect-offset-commit

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

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


commit 4d1db91ec8276da3dcca6d85b66c70497cf0815d
Author: simplesteph 
Date:   2017-03-24T00:03:07Z

fixed the commitTimeoutMs being set before the commit actually started
this fixes KAFKA-4942




> Kafka Connect: Offset committing times out before expected
> --
>
> Key: KAFKA-4942
> URL: https://issues.apache.org/jira/browse/KAFKA-4942
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.2.0
>Reporter: Stephane Maarek
>
> On Kafka 0.10.2.0
> I run a connector that deals with a lot of data, in a kafka connect cluster
> When the offsets are getting committed, I get the following:
> {code}
> [2017-03-23 03:56:25,134] INFO WorkerSinkTask{id=MyConnector-1} Committing 
> offsets (org.apache.kafka.connect.runtime.WorkerSinkTask)
> [2017-03-23 03:56:25,135] WARN Commit of WorkerSinkTask{id=MyConnector-1} 
> offsets timed out (org.apache.kafka.connect.runtime.WorkerSinkTask)
> {code}
> If you look at the timestamps, they're 1 ms apart. My settings are the 
> following: 
> {code}
>   offset.flush.interval.ms = 12
>   offset.flush.timeout.ms = 6
>   offset.storage.topic = _connect_offsets
> {code}
> It seems the offset flush timeout setting is completely ignored for the look 
> of the logs. I would expect the timeout message to happen 60 seconds after 
> the commit offset INFO message, not 1 millisecond later.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [VOTE] KIP-82 Add Record Headers

[Jun Rao]

Hi, Michael,

The KIP looks good to me overall. Just one comment. The wiki says "This
will be done by calling "close()" method". However, there is no close() in
Headers.

Thanks,

Jun

On Thu, Mar 23, 2017 at 9:34 AM, Michael Pearce 
wrote:

> Thanks all for the +1 votes so far, just one more committer vote needed (
>
> Please note:
>
> I have made one small adjustment to the kip based on Ismael’s comment in
> discussion thread, and further requested by Jason in the vote thread.
>
> Please note the below method is changed based on this feedback.
>
> Headers.header(key)` to be `Headers.lastHeader(key)
>
> Thanks
> Mike
>
> On 22/03/2017, 16:39, "Joel Koshy"  wrote:
>
> +1
>
> On Tue, Mar 21, 2017 at 5:01 PM, Jason Gustafson 
> wrote:
>
> > Thanks for the KIP! +1 (binding) from me. Just one nit: can we change
> > `Headers.header(key)` to be `Headers.lastHeader(key)`? It's not a
> > deal-breaker, but I think it's better to let the name reflect the
> actual
> > behavior as clearly as possible.
> >
> > -Jason
> >
> > On Wed, Feb 15, 2017 at 6:10 AM, Jeroen van Disseldorp <
> jer...@axual.io>
> > wrote:
> >
> > > +1 on introducing the concept of headers, neutral on specific
> > > implementation.
> > >
> > >
> > >
> > > On 14/02/2017 22:34, Jay Kreps wrote:
> > >
> > >> Couple of things I think we still need to work out:
> > >>
> > >> 1. I think we agree about the key, but I think we haven't
> talked
> > about
> > >> the value yet. I think if our goal is an open ecosystem of
> these
> > >> header
> > >> spread across many plugins from many systems we should
> consider
> > >> making this
> > >> a string as well so it can be printed, set via a UI, set in
> config,
> > >> etc.
> > >> Basically encouraging pluggable serialization formats here
> will lead
> > >> to a
> > >> bit of a tower of babel.
> > >> 2. This proposal still includes a pretty big change to our
> > >> serialization
> > >> and protocol definition layer. Essentially it is introducing
> an
> > >> optional
> > >> type, where the format is data dependent. I think this is
> actually a
> > >> big
> > >> change though it doesn't seem like it. It means you can no
> longer
> > >> specify
> > >> this type with our type definition DSL, and likewise it
> requires
> > >> custom
> > >> handling in client libs. This isn't a huge thing, since the
> Record
> > >> definition is custom anyway, but I think this kind of protocol
> > >> inconsistency is very non-desirable and ties you to
> hand-coding
> > >> things. I
> > >> think the type should instead by [Key Value] in our BNF,
> where key
> > and
> > >> value are both short strings as used elsewhere. This brings
> it in
> > >> line with
> > >> the rest of the protocol.
> > >> 3. Could we get more specific about the exact Java API change
> to
> > >> ProducerRecord, ConsumerRecord, Record, etc?
> > >>
> > >> -Jay
> > >>
> > >> On Tue, Feb 14, 2017 at 9:42 AM, Michael Pearce <
> michael.pea...@ig.com>
> > >> wrote:
> > >>
> > >> Hi all,
> > >>>
> > >>> We would like to start the voting process for KIP-82 – Add record
> > >>> headers.
> > >>> The KIP can be found
> > >>> at
> > >>>
> > >>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > >>> 82+-+Add+Record+Headers
> > >>>
> > >>> Discussion thread(s) can be found here:
> > >>>
> > >>> http://search-hadoop.com/m/Kafka/uyzND1nSTOHTvj81?subj=
> > >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> > >>> http://search-hadoop.com/m/Kafka/uyzND1Arxt22Tvj81?subj=
> > >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> > >>> http://search-hadoop.com/?project=Kafka&q=KIP-82
> > >>>
> > >>>
> > >>>
> > >>> Thanks,
> > >>> Mike
> > >>>
> > >>> The information contained in this email is strictly confidential
> and
> > for
> > >>> the use of the addressee only, unless otherwise indicated. If
> you are
> > not
> > >>> the intended recipient, please do not read, copy, use or
> disclose to
> > >>> others
> > >>> this message or any attachment. Please also notify the sender by
> > replying
> > >>> to this email or by telephone (+44(020 7896 0011) and then
> delete the
> > >>> email
> > >>> and any copies of it. Opinions, conclusion (etc) that do not
> relate to
> > >>> the
> > >>> official business of this company shall be understood as neither
> given
> > >>> nor
> > >>> endorsed by it. IG is a trading name of IG Markets Limited (a
> company
> > >>> registered in England and Wales, company number 04008957) and IG
> Index
> > >>> Limited (a company registered in England and Wales, company
> number
> > >>> 01190902). Registered a

Re: [VOTE] KIP-82 Add Record Headers

[Ismael Juma]

Hi Jun,

The close method will only exist in the implementation class as it's not
meant to be called from user code.

Ismael

On 24 Mar 2017 1:51 am, "Jun Rao"  wrote:

> Hi, Michael,
>
> The KIP looks good to me overall. Just one comment. The wiki says "This
> will be done by calling "close()" method". However, there is no close() in
> Headers.
>
> Thanks,
>
> Jun
>
> On Thu, Mar 23, 2017 at 9:34 AM, Michael Pearce 
> wrote:
>
> > Thanks all for the +1 votes so far, just one more committer vote needed (
> >
> > Please note:
> >
> > I have made one small adjustment to the kip based on Ismael’s comment in
> > discussion thread, and further requested by Jason in the vote thread.
> >
> > Please note the below method is changed based on this feedback.
> >
> > Headers.header(key)` to be `Headers.lastHeader(key)
> >
> > Thanks
> > Mike
> >
> > On 22/03/2017, 16:39, "Joel Koshy"  wrote:
> >
> > +1
> >
> > On Tue, Mar 21, 2017 at 5:01 PM, Jason Gustafson  >
> > wrote:
> >
> > > Thanks for the KIP! +1 (binding) from me. Just one nit: can we
> change
> > > `Headers.header(key)` to be `Headers.lastHeader(key)`? It's not a
> > > deal-breaker, but I think it's better to let the name reflect the
> > actual
> > > behavior as clearly as possible.
> > >
> > > -Jason
> > >
> > > On Wed, Feb 15, 2017 at 6:10 AM, Jeroen van Disseldorp <
> > jer...@axual.io>
> > > wrote:
> > >
> > > > +1 on introducing the concept of headers, neutral on specific
> > > > implementation.
> > > >
> > > >
> > > >
> > > > On 14/02/2017 22:34, Jay Kreps wrote:
> > > >
> > > >> Couple of things I think we still need to work out:
> > > >>
> > > >> 1. I think we agree about the key, but I think we haven't
> > talked
> > > about
> > > >> the value yet. I think if our goal is an open ecosystem of
> > these
> > > >> header
> > > >> spread across many plugins from many systems we should
> > consider
> > > >> making this
> > > >> a string as well so it can be printed, set via a UI, set in
> > config,
> > > >> etc.
> > > >> Basically encouraging pluggable serialization formats here
> > will lead
> > > >> to a
> > > >> bit of a tower of babel.
> > > >> 2. This proposal still includes a pretty big change to our
> > > >> serialization
> > > >> and protocol definition layer. Essentially it is introducing
> > an
> > > >> optional
> > > >> type, where the format is data dependent. I think this is
> > actually a
> > > >> big
> > > >> change though it doesn't seem like it. It means you can no
> > longer
> > > >> specify
> > > >> this type with our type definition DSL, and likewise it
> > requires
> > > >> custom
> > > >> handling in client libs. This isn't a huge thing, since the
> > Record
> > > >> definition is custom anyway, but I think this kind of
> protocol
> > > >> inconsistency is very non-desirable and ties you to
> > hand-coding
> > > >> things. I
> > > >> think the type should instead by [Key Value] in our BNF,
> > where key
> > > and
> > > >> value are both short strings as used elsewhere. This brings
> > it in
> > > >> line with
> > > >> the rest of the protocol.
> > > >> 3. Could we get more specific about the exact Java API
> change
> > to
> > > >> ProducerRecord, ConsumerRecord, Record, etc?
> > > >>
> > > >> -Jay
> > > >>
> > > >> On Tue, Feb 14, 2017 at 9:42 AM, Michael Pearce <
> > michael.pea...@ig.com>
> > > >> wrote:
> > > >>
> > > >> Hi all,
> > > >>>
> > > >>> We would like to start the voting process for KIP-82 – Add
> record
> > > >>> headers.
> > > >>> The KIP can be found
> > > >>> at
> > > >>>
> > > >>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > > >>> 82+-+Add+Record+Headers
> > > >>>
> > > >>> Discussion thread(s) can be found here:
> > > >>>
> > > >>> http://search-hadoop.com/m/Kafka/uyzND1nSTOHTvj81?subj=
> > > >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> > > >>> http://search-hadoop.com/m/Kafka/uyzND1Arxt22Tvj81?subj=
> > > >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> > > >>> http://search-hadoop.com/?project=Kafka&q=KIP-82
> > > >>>
> > > >>>
> > > >>>
> > > >>> Thanks,
> > > >>> Mike
> > > >>>
> > > >>> The information contained in this email is strictly
> confidential
> > and
> > > for
> > > >>> the use of the addressee only, unless otherwise indicated. If
> > you are
> > > not
> > > >>> the intended recipient, please do not read, copy, use or
> > disclose to
> > > >>> others
> > > >>> this message or any attachment. Please also notify the sender
> by
> > > replying
> > > >>> to this email or by telephone (+44(020 7896 0011) and then
> > delete the
> > > >>> email
> > > >>> and a

Re: [VOTE] KIP-123: Allow per stream/table timestamp extractor

[Matthias J. Sax]

Thanks Jeyhun.

Can you also update the KIP accordingly. It must contain all changes to
public API. Thus, list all parameters that get deprecated and newly
added. And add a sentence about backward compatibility.


-Matthias

On 3/23/17 3:16 AM, Jeyhun Karimov wrote:
> Sorry for a super late update. I made an update on related PR.
> 
> Cheers,
> Jeyhun
> 
> On Wed, Mar 22, 2017 at 9:09 PM Guozhang Wang  wrote:
> 
>> Jeyhun,
>>
>> Could you update the status of this KIP since it has been some time since
>> the last vote?
>>
>> I'm +1 besides the minor comments mentioned above.
>>
>>
>> Guozhang
>>
>>
>> On Mon, Mar 6, 2017 at 3:14 PM, Jeyhun Karimov 
>> wrote:
>>
>>> Sorry I was late. I will update javadocs in related methods to emphasize
>>> that TimestampExtractor is stateless.
>>>
>>> Cheers,
>>> Jeyhun
>>>
>>> On Mon, Mar 6, 2017 at 8:17 PM Guozhang Wang  wrote:
>>>
 1) Sounds good.

 2) Yeah what I meant is to emphasize that TimestampExtractor to be
 stateless in the docs somewhere.


 Guozhang


 On Sun, Mar 5, 2017 at 4:27 PM, Matthias J. Sax >>
 wrote:

> Guozhang,
>
> about renaming the config parameters. I like this idea, but want to
> point out, that this change should be done in a backward compatible
>>> way.
> Thus, we need to keep (and only deprecate) the current parameter
>> names.
>
> I am not sure about (2)? What do you worry about? Using a "stateful
> extractor"? -- this would be an antipattern IMHO. We can clarify
>> that a
> TimestampExtrator should be stateless though (even if this should be
> clear).
>
>
> -Matthias
>
>
> On 3/4/17 6:36 PM, Guozhang Wang wrote:
>> Jeyhun,
>>
>> Thanks for proposing this KIP! And sorry for getting late in the
> discussion.
>>
>> I have a general suggestion not directly related to this KIP and a
 couple
>> of comments for this KIP here:
>>
>> I agree with Mathieu's observation, partly because we are now
>> having
 lots
>> of overloaded functions both in the DSL and in PAPI, and it would
>> be
> quite
>> confusing to users. As Matthias mentioned we do have some plans to
> refactor
>> this API, but just wanted to point it out that this KIP may likely
>>> urge
> us
>> to do the API refactoring sooner than planned. My personal
>> preference
> would
>> be doing that the next release (i.e. 0.11.0.0 in June).
>>
>>
>> Now some detailed comments:
>>
>> 1. I'd suggest change TIMESTAMP_EXTRACTOR_CLASS_CONFIG to
>> "default.timestamp.extractor" or "global.timestamp.extractor" (also
>>> the
>> Java variable name can be changed accordingly) along with this
>>> change.
 In
>> addition, maybe we can piggy-backing this to also rename
>> KEY_SERDE_CLASS_CONFIG/VALUE_SERDE_CLASS_CONFIG to "default.key.."
>>> etc
> in
>> this KIP.
>>
>> 2. Another thing we should consider, is that since now we could
> potentially
>> use multiple timestamp extractor instances than a single one, this
>>> may
 be
>> breaking if user's customization did some global bookkeeping based
>> on
 the
>> previous assumption (maybe a wild thought but e.g. keeping track
>> some
>> global counts in the extractor as a local variable). We need to
>>> clarify
>> this change in the javadoc and also potentially in the upgrade web
>>> doc
>> sections.
>>
>>
>>
>> Guozhang
>>
>>
>> On Wed, Mar 1, 2017 at 6:09 AM, Michael Noll >>
> wrote:
>>
>>> +1 (non-binding)
>>>
>>> Thanks for the KIP!
>>>
>>> On Wed, Mar 1, 2017 at 1:49 PM, Bill Bejeck 
 wrote:
>>>
 +1

 Thanks
 Bill

 On Wed, Mar 1, 2017 at 5:06 AM, Eno Thereska <
>>> eno.there...@gmail.com
>
 wrote:

> +1 (non binding).
>
> Thanks
> Eno
>> On 28 Feb 2017, at 17:22, Matthias J. Sax <
>> matth...@confluent.io

 wrote:
>>
>> +1
>>
>> Thanks a lot for the KIP!
>>
>> -Matthias
>>
>>
>> On 2/28/17 1:35 AM, Damian Guy wrote:
>>> Thanks for the KIP Jeyhun!
>>>
>>> +1
>>>
>>> On Tue, 28 Feb 2017 at 08:59 Jeyhun Karimov <
>>> je.kari...@gmail.com
>
> wrote:
>>>
 Dear community,

 I'd like to start the vote for KIP-123:
 https://cwiki.apache.org/confluence/pages/viewpage.
> action?pageId=68714788


 Cheers,
 Jeyhun
 --
 -Cheers

 Jeyhun

>>>
>>
>
>

>>>
>>
>>
>>
>
>


 --
 -- Guozhang

>>> --
>>> -Cheers
>>>
>>> Jeyhun
>>>
>>
>>
>>
>> 

Re: [VOTE] KIP-82 Add Record Headers

[Jun Rao]

Hi, Ismael,

Ok, that make sense.


Hi, Michael,

Could we document this in the wiki?

+1 from me on the KIP.

Thanks,

Jun

On Thu, Mar 23, 2017 at 7:58 PM, Ismael Juma  wrote:

> Hi Jun,
>
> The close method will only exist in the implementation class as it's not
> meant to be called from user code.
>
> Ismael
>
> On 24 Mar 2017 1:51 am, "Jun Rao"  wrote:
>
> > Hi, Michael,
> >
> > The KIP looks good to me overall. Just one comment. The wiki says "This
> > will be done by calling "close()" method". However, there is no close()
> in
> > Headers.
> >
> > Thanks,
> >
> > Jun
> >
> > On Thu, Mar 23, 2017 at 9:34 AM, Michael Pearce 
> > wrote:
> >
> > > Thanks all for the +1 votes so far, just one more committer vote
> needed (
> > >
> > > Please note:
> > >
> > > I have made one small adjustment to the kip based on Ismael’s comment
> in
> > > discussion thread, and further requested by Jason in the vote thread.
> > >
> > > Please note the below method is changed based on this feedback.
> > >
> > > Headers.header(key)` to be `Headers.lastHeader(key)
> > >
> > > Thanks
> > > Mike
> > >
> > > On 22/03/2017, 16:39, "Joel Koshy"  wrote:
> > >
> > > +1
> > >
> > > On Tue, Mar 21, 2017 at 5:01 PM, Jason Gustafson <
> ja...@confluent.io
> > >
> > > wrote:
> > >
> > > > Thanks for the KIP! +1 (binding) from me. Just one nit: can we
> > change
> > > > `Headers.header(key)` to be `Headers.lastHeader(key)`? It's not a
> > > > deal-breaker, but I think it's better to let the name reflect the
> > > actual
> > > > behavior as clearly as possible.
> > > >
> > > > -Jason
> > > >
> > > > On Wed, Feb 15, 2017 at 6:10 AM, Jeroen van Disseldorp <
> > > jer...@axual.io>
> > > > wrote:
> > > >
> > > > > +1 on introducing the concept of headers, neutral on specific
> > > > > implementation.
> > > > >
> > > > >
> > > > >
> > > > > On 14/02/2017 22:34, Jay Kreps wrote:
> > > > >
> > > > >> Couple of things I think we still need to work out:
> > > > >>
> > > > >> 1. I think we agree about the key, but I think we haven't
> > > talked
> > > > about
> > > > >> the value yet. I think if our goal is an open ecosystem of
> > > these
> > > > >> header
> > > > >> spread across many plugins from many systems we should
> > > consider
> > > > >> making this
> > > > >> a string as well so it can be printed, set via a UI, set
> in
> > > config,
> > > > >> etc.
> > > > >> Basically encouraging pluggable serialization formats here
> > > will lead
> > > > >> to a
> > > > >> bit of a tower of babel.
> > > > >> 2. This proposal still includes a pretty big change to our
> > > > >> serialization
> > > > >> and protocol definition layer. Essentially it is
> introducing
> > > an
> > > > >> optional
> > > > >> type, where the format is data dependent. I think this is
> > > actually a
> > > > >> big
> > > > >> change though it doesn't seem like it. It means you can no
> > > longer
> > > > >> specify
> > > > >> this type with our type definition DSL, and likewise it
> > > requires
> > > > >> custom
> > > > >> handling in client libs. This isn't a huge thing, since
> the
> > > Record
> > > > >> definition is custom anyway, but I think this kind of
> > protocol
> > > > >> inconsistency is very non-desirable and ties you to
> > > hand-coding
> > > > >> things. I
> > > > >> think the type should instead by [Key Value] in our BNF,
> > > where key
> > > > and
> > > > >> value are both short strings as used elsewhere. This
> brings
> > > it in
> > > > >> line with
> > > > >> the rest of the protocol.
> > > > >> 3. Could we get more specific about the exact Java API
> > change
> > > to
> > > > >> ProducerRecord, ConsumerRecord, Record, etc?
> > > > >>
> > > > >> -Jay
> > > > >>
> > > > >> On Tue, Feb 14, 2017 at 9:42 AM, Michael Pearce <
> > > michael.pea...@ig.com>
> > > > >> wrote:
> > > > >>
> > > > >> Hi all,
> > > > >>>
> > > > >>> We would like to start the voting process for KIP-82 – Add
> > record
> > > > >>> headers.
> > > > >>> The KIP can be found
> > > > >>> at
> > > > >>>
> > > > >>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > > > >>> 82+-+Add+Record+Headers
> > > > >>>
> > > > >>> Discussion thread(s) can be found here:
> > > > >>>
> > > > >>> http://search-hadoop.com/m/Kafka/uyzND1nSTOHTvj81?subj=
> > > > >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> > > > >>> http://search-hadoop.com/m/Kafka/uyzND1Arxt22Tvj81?subj=
> > > > >>> Re+DISCUSS+KIP+82+Add+Record+Headers
> > > > >>> http://search-hadoop.com/?project=Kafka&q=KIP-82
> > > > >>>
> > > > >>>
> > > > >>>
> > > > >>> Thanks,
> > > > >>> Mike
> > > > >>>
> > > > >>> The information conta

Re: [DISCUSS] KIP-113: Support replicas movement between log directories

[Jun Rao]

Hi, Dong,

11.2 I think there are a few reasons why the cross disk movement may not
catch up if the replicas are created in the wrong log dirs to start with.
(a) There could be more replica fetcher threads than the disk movement
threads. (b) intra.broker.throttled.rate may be configured lower than the
replica throttle rate. That's why I think getting the replicas created in
the right log dirs will be better.

For the corner case issue that you mentioned, I am not sure if the approach
in the KIP completely avoids that. If a broker is down when the partition
reassignment tool is started, does the tool just hang (keep retrying
ChangeReplicaDirRequest) until the broker comes back? Currently, the
partition reassignment tool doesn't block.

Thanks,

Jun


On Tue, Mar 21, 2017 at 11:24 AM, Dong Lin  wrote:

> Hey Jun,
>
> Thanks for the explanation. Please see below my thoughts.
>
> 10. I see. So you are concerned with the potential implementation
> complexity which I wasn't aware of. I think it is OK not to do log
> cleaning on the .move log since there can be only one such log in each
> directory. I have updated the KIP to specify this:
>
> "The log segments in topicPartition.move directory will be subject to log
> truncation, log retention in the same way as the log segments in the source
> log directory. But we may not do log cleaning on the topicPartition.move to
> simplify the implementation."
>
> 11.2 Now I get your point. I think we have slightly different expectation
> of the order in which the reassignment tools updates reassignment node in
> ZK and sends ChangeReplicaDirRequest.
>
> I think the reassignment tool should first create reassignment znode and
> then keep sending ChangeReplicaDirRequest until success. I think sending
> ChangeReplicaDirRequest before updating znode has negligible impact on the
> chance that the broker processes ChangeReplicaDirRequest before
> LeaderAndIsrRequest from controller, because the time for controller to
> receive ZK notification, handle state machine changes and send
> LeaderAndIsrRequests should be much longer than the time for reassignment
> tool to setup connection with broker and send ChangeReplicaDirRequest. Even
> if broker receives LeaderAndIsrRequest a bit sooner, the data in the
> original replica should be smaller enough for .move log to catch up very
> quickly, so that broker can swap the log soon after it receives
> ChangeReplicaDirRequest -- otherwise the intra.broker.throttled.rate is
> probably too small. Does this address your concern with the performance?
>
> One concern with the suggested approach is that the ChangeReplicaDirRequest
> may be lost if broker crashes before it creates the replica. I agree it is
> rare. But it will be confusing when it happens. Operators would have to
> keep verifying reassignment and possibly retry execution until success if
> they want to make sure that the ChangeReplicaDirRequest is executed.
>
> Thanks,
> Dong
>
>
>
> On Tue, Mar 21, 2017 at 8:37 AM, Jun Rao  wrote:
>
> > Hi, Dong,
> >
> > 10. I was mainly concerned about the additional complexity needed to
> > support log cleaning in the .move log. For example, LogToClean is keyed
> off
> > TopicPartition. To be able to support cleaning different instances of the
> > same partition, we need additional logic. I am not how much additional
> > complexity is needed and whether it's worth it. If we don't do log
> cleaning
> > at all on the .move log, then we don't have to change the log cleaner's
> > code.
> >
> > 11.2 I was thinking of the following flow. In the execute phase, the
> > reassignment tool first issues a ChangeReplicaDirRequest to brokers where
> > new replicas will be created. The brokers remember the mapping and
> return a
> > successful code. The reassignment tool then initiates the cross broker
> > movement through the controller. In the verify phase, in addition to
> > checking the replica assignment at the brokers, it issues
> > DescribeDirsRequest to check the replica to log dirs mapping. For each
> > partition in the response, the broker returns a state to indicate whether
> > the replica is final, temporary or pending. If all replicas are in the
> > final state, the tool checks if all replicas are in the expected log
> dirs.
> > If they are not, output a warning (and perhaps suggest the users to move
> > the data again). However, this should be rare.
> >
> > Thanks,
> >
> > Jun
> >
> >
> > On Mon, Mar 20, 2017 at 10:46 AM, Dong Lin  wrote:
> >
> > > Hey Jun,
> > >
> > > Thanks for the response! It seems that we have only two remaining
> issues.
> > > Please see my reply below.
> > >
> > > On Mon, Mar 20, 2017 at 7:45 AM, Jun Rao  wrote:
> > >
> > > > Hi, Dong,
> > > >
> > > > Thanks for the update. A few replies inlined below.
> > > >
> > > > On Thu, Mar 16, 2017 at 12:28 AM, Dong Lin 
> > wrote:
> > > >
> > > > > Hey Jun,
> > > > >
> > > > > Thanks for your comment! Please see my reply below.
> > > > >
> > > > > On Wed, Mar 15, 2017 at 9:45 PM,

Re: [DISCUSS] KIP-134: Delay initial consumer group rebalance

[Ewen Cheslack-Postava]

I have the same initial response as Ismael re: broker vs consumer settings.
The global setting seems questionable.

Could we maybe summarize what the impact of making this a client config
would be? Protocol bump is obvious, but is there any other significant
issue? For the protocol bump in particular, I think this change is
currently really critical for streams; it will be valuable elsewhere, but
the immediate demand is streams, so a protocol bump while being backwards
compatible wouldn't affect any other clients. Is this still actually
compatible with different clients given that they would now expect
different timeouts? (I think it's strictly compatible if you wait for
responses, but if you enforce any client side timeouts, I'm not so sure.)

re: test plan, I'm sure this will come as a surprise, but is the system
test even necessary? Validating # of rebalances seems messy as other things
can cause rebalances (though admittedly not in a "clean" case). But really
it seems like an integration test could validate this by making sure only 1
rebalance occurred when 2 members joined with a sufficient time gap.

-Ewen

On Thu, Mar 23, 2017 at 3:53 PM, Matthias J. Sax 
wrote:

> Thanks for the KIP Damian!
>
> My two cents:
>
>  - we should have an explicit parameter for this -- implicit setting are
> always tricky (the "importance" of this parameter would be LOW)
>
>  - the config should be different for each consumer group:
>* assume you have a stateless app, you want to rebalance immediately
>* if you start-up in an visualized environment using some tools like
> Mesos you might need a different value that on bare metal (no VM to be
> started)
>* it also depends, how many consumer instanced you expect -- it's
> harder to start up 100 instances in 3 seconds than 5
>
>  - the default value should be zero
>
>
> One more thought: what about scaling scenarios? If a consumer group has
> 10 instanced and should be scaled up to 20, it would make sense to do
> this with a single rebalance, too. Thus, I am wondering, if it would
> make sense to apply this delay each time a new consumer joins group,
> even if the group is not empty?
>
>
> -Matthias
>
>
> On 3/23/17 10:19 AM, Damian Guy wrote:
> > Thanks Gouzhang - i think another problem with this is that is
> overloading
> > session.timeout.ms to mean multiple things. I'm not sure that is a good
> > thing.
> >
> > On Thu, 23 Mar 2017 at 17:14 Guozhang Wang  wrote:
> >
> >> The downside of it, though, is that although it "hides" this from most
> of
> >> the users needing to be aware of it, by default session timeout i.e. the
> >> rebalance timeout is 10 seconds which could arguably too long.
> >>
> >>
> >> Guozhang
> >>
> >> On Thu, Mar 23, 2017 at 10:12 AM, Guozhang Wang 
> >> wrote:
> >>
> >>> Just throwing another alternative idea here: we can consider using the
> >>> rebalance timeout value which is already included in the join request
> >>> protocol (and on the current Java client it is always written as the
> >>> session timeout value), that the first member joining will always force
> >> the
> >>> coordinator to wait that long. By doing this we do not need to bump up
> >> the
> >>> protocol either.
> >>>
> >>>
> >>> Guozhang
> >>>
> >>> On Thu, Mar 23, 2017 at 5:49 AM, Damian Guy 
> >> wrote:
> >>>
>  Hi Ismael,
> 
>  Mostly to avoid the protocol bump.
> 
>  I agree that it may be difficult to choose the right delay for all
>  consumer
>  groups, but we wanted to make this something that most users don't
> >> really
>  need to think about, i.e., a small enough default delay that works in
> >> the
>  majority of cases. However it would be much more flexible as a
> consumer
>  config, which i'm happy to pursue if this change is worthy of a
> protocol
>  bump.
> 
>  Thanks,
>  Damian
> 
>  On Thu, 23 Mar 2017 at 12:35 Ismael Juma  wrote:
> 
> > Thanks for the KIP, Damian. It makes sense to avoid multiple
> >> rebalances
> > during start-up. One issue with having this as a broker config is
> that
>  it
> > may be difficult to choose the right delay for all consumer groups.
> >> Can
>  you
> > elaborate a little more on why the first alternative (add a consumer
> > config) was rejected? We bump protocol versions regularly (when it
> >> makes
> > sense), so it would be good to get a bit more detail.
> >
> > Thanks,
> > Ismael
> >
> > On Thu, Mar 23, 2017 at 12:24 PM, Damian Guy 
>  wrote:
> >
> >> Hi All,
> >>
> >> I've prepared a KIP to add a configurable delay to the initial
>  consumer
> >> group rebalance.
> >>
> >> Please have look here:
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> >> 134%3A+Delay+initial+consumer+group+rebalance
> >>
> >> Thanks,
> >> Damian
> >>
> >> BTW, i apologize if this appears twice. Seems the first one may have
>  not
> >> m

[jira] [Commented] (KAFKA-4943) SCRAM secret's should be better protected with Zookeeper ACLs

[JIRA]

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

Johan Ström commented on KAFKA-4943:


I agree, it won't be totally secure unless Zk is totally secure, but this more 
or less requires that the Zk instance is dedicated to only Kafka in order to be 
"secure". In my current setup I have a "secure" Zk, in terms of private 
networks etc, but there are still other (legit) applications which have access 
to their parts of the Zookeeper tree. No need for them to be able to snoop on 
all SCRAM credentials, even if there are some other potential attack vectors if 
they happen to be on the network close to the brokers/zk servers.

So yes, looking forward to some tighter ACLs :)
If not (or until fixed), this should be mentioned in the docs, as they 
currently have the following phrase:
"The metadata stored in ZooKeeper for the Kafka cluster is world-readable, but 
can only be modified by the brokers. The rationale behind this decision is that 
the data stored in ZooKeeper is not sensitive, but inappropriate manipulation 
of that data can cause cluster disruption."

> SCRAM secret's should be better protected with Zookeeper ACLs
> -
>
> Key: KAFKA-4943
> URL: https://issues.apache.org/jira/browse/KAFKA-4943
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Johan Ström
>
> With the new SCRAM authenticator the secrets are stored in Zookeeper:
> {code}
> get /kafka/config/users/alice
> {"version":1,"config":{"SCRAM-SHA-512":"salt=ODhnZjNkdWZibTV1cG1zdnV6bmh6djF3Mg==,stored_key=BAbHWHuGEb4m5+U+p0M9oFQmOPhU6M7q5jtZY8deDDoZCvxaqVNLz41yPzdgcp1WpiEBmfwYOuFlo9hMFKM7mA==,server_key=JW3KhpMeyUgh0OAC0kejuFUvUSlXBv/Z68tlfOWcMw5f5jrBwyBnjNQ9VZsSYz1AcI9IYaQ5S6H3yN39SieNiA==,iterations=4096"}}
> {code}
> These are stored without any ACL, and zookeeper-security-migration.sh does 
> not seem to change that either:
> {code}
> getAcl /kafka/config/users/alice
> 'world,'anyone
> : cdrwa
> getAcl /kafka/config/users
> 'world,'anyone
> : cdrwa
> getAcl /kafka
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> getAcl /kafka/config/changes
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> {code}
> The above output is after running security migrator, for some reason 
> /kafka/config/users is ignored, but others are fixed..
> Even if these where to be stored with secure ZkUtils#DefaultAcls, they would 
> be world readable.
> From my (limited) point of view, they should be readable by Kafka only.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4943) SCRAM secret's should be better protected with Zookeeper ACLs

[Stephane Maarek (JIRA)]

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

Stephane Maarek commented on KAFKA-4943:


[~rsivaram] if Zookeeper has only cdrwa by the kafka broker, and just no world 
: r permission, wouldn't that fix any concerns with a world-opened zookeeper?

> SCRAM secret's should be better protected with Zookeeper ACLs
> -
>
> Key: KAFKA-4943
> URL: https://issues.apache.org/jira/browse/KAFKA-4943
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Johan Ström
>
> With the new SCRAM authenticator the secrets are stored in Zookeeper:
> {code}
> get /kafka/config/users/alice
> {"version":1,"config":{"SCRAM-SHA-512":"salt=ODhnZjNkdWZibTV1cG1zdnV6bmh6djF3Mg==,stored_key=BAbHWHuGEb4m5+U+p0M9oFQmOPhU6M7q5jtZY8deDDoZCvxaqVNLz41yPzdgcp1WpiEBmfwYOuFlo9hMFKM7mA==,server_key=JW3KhpMeyUgh0OAC0kejuFUvUSlXBv/Z68tlfOWcMw5f5jrBwyBnjNQ9VZsSYz1AcI9IYaQ5S6H3yN39SieNiA==,iterations=4096"}}
> {code}
> These are stored without any ACL, and zookeeper-security-migration.sh does 
> not seem to change that either:
> {code}
> getAcl /kafka/config/users/alice
> 'world,'anyone
> : cdrwa
> getAcl /kafka/config/users
> 'world,'anyone
> : cdrwa
> getAcl /kafka
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> getAcl /kafka/config/changes
> 'world,'anyone
> : r
> 'sasl,'bob
> : cdrwa
> {code}
> The above output is after running security migrator, for some reason 
> /kafka/config/users is ignored, but others are fixed..
> Even if these where to be stored with secure ZkUtils#DefaultAcls, they would 
> be world readable.
> From my (limited) point of view, they should be readable by Kafka only.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: [DISCUSS] KIP-113: Support replicas movement between log directories

[Dong Lin]

Hey Jun,

Thanks much for the response! I agree with you that if multiple replicas
are created in the wrong directory, we may waste resource if either
replicaMoveThread number is low or intra.broker.throttled.rate is slow.
Then the question is whether the suggested approach increases the chance of
replica being created in the correct log directory.

I think the answer is no due to the argument provided in the previous
email. Sending ChangeReplicaDirRequest before updating znode has negligible
impact on the chance that the broker processes ChangeReplicaDirRequest
before LeaderAndIsrRequest from controller. If we still worry about the
order they are sent, the reassignment tool can first send
ChangeReplicaDirRequest (so that broker remembers it in memory), create
reassignment znode, and then retry ChangeReplicaDirRequset if the previous
ChangeReplicaDirResponse says the replica has not been created. This should
give us the highest possible chance of creating replica in the correct
directory and avoid the problem of the suggested approach. I have updated "How
to reassign replica between log directories across brokers" in the KIP to
explain this procedure.

To answer your question, the reassignment tool should fail with with proper
error message if user has specified log directory for a replica on an
offline broker.  This is reasonable because reassignment tool can not
guarantee that the replica will be moved to the specified log directory if
the broker is offline. If all brokers are online, the reassignment tool may
hung up to 10 seconds (by default) to retry ChangeReplicaDirRequest if any
replica has not been created already. User can change this timeout value
using the newly-added --timeout argument of the reassignment tool. This is
specified in the Public Interface section in the KIP. The reassignment tool
will only block if user uses this new feature of reassigning replica to a
specific log directory in the broker. Therefore it seems backward
compatible.

Does this address the concern?

Thanks,
Dong

On Thu, Mar 23, 2017 at 10:06 PM, Jun Rao  wrote:

> Hi, Dong,
>
> 11.2 I think there are a few reasons why the cross disk movement may not
> catch up if the replicas are created in the wrong log dirs to start with.
> (a) There could be more replica fetcher threads than the disk movement
> threads. (b) intra.broker.throttled.rate may be configured lower than the
> replica throttle rate. That's why I think getting the replicas created in
> the right log dirs will be better.
>
> For the corner case issue that you mentioned, I am not sure if the approach
> in the KIP completely avoids that. If a broker is down when the partition
> reassignment tool is started, does the tool just hang (keep retrying
> ChangeReplicaDirRequest) until the broker comes back? Currently, the
> partition reassignment tool doesn't block.
>
> Thanks,
>
> Jun
>
>
> On Tue, Mar 21, 2017 at 11:24 AM, Dong Lin  wrote:
>
> > Hey Jun,
> >
> > Thanks for the explanation. Please see below my thoughts.
> >
> > 10. I see. So you are concerned with the potential implementation
> > complexity which I wasn't aware of. I think it is OK not to do log
> > cleaning on the .move log since there can be only one such log in each
> > directory. I have updated the KIP to specify this:
> >
> > "The log segments in topicPartition.move directory will be subject to log
> > truncation, log retention in the same way as the log segments in the
> source
> > log directory. But we may not do log cleaning on the topicPartition.move
> to
> > simplify the implementation."
> >
> > 11.2 Now I get your point. I think we have slightly different expectation
> > of the order in which the reassignment tools updates reassignment node in
> > ZK and sends ChangeReplicaDirRequest.
> >
> > I think the reassignment tool should first create reassignment znode and
> > then keep sending ChangeReplicaDirRequest until success. I think sending
> > ChangeReplicaDirRequest before updating znode has negligible impact on
> the
> > chance that the broker processes ChangeReplicaDirRequest before
> > LeaderAndIsrRequest from controller, because the time for controller to
> > receive ZK notification, handle state machine changes and send
> > LeaderAndIsrRequests should be much longer than the time for reassignment
> > tool to setup connection with broker and send ChangeReplicaDirRequest.
> Even
> > if broker receives LeaderAndIsrRequest a bit sooner, the data in the
> > original replica should be smaller enough for .move log to catch up very
> > quickly, so that broker can swap the log soon after it receives
> > ChangeReplicaDirRequest -- otherwise the intra.broker.throttled.rate is
> > probably too small. Does this address your concern with the performance?
> >
> > One concern with the suggested approach is that the
> ChangeReplicaDirRequest
> > may be lost if broker crashes before it creates the replica. I agree it
> is
> > rare. But it will be confusing when it happens. Operators would have to
> >