kafka.git
3 min agoMINOR: additional check to follower fetch handling (#4433) trunk
Edoardo Comar [Thu, 18 Jan 2018 19:14:49 +0000 (19:14 +0000)] 
MINOR: additional check to follower fetch handling  (#4433)

add check to KafkaApis, add unit test specific to follower fetch
developed with @mimaison

Reviewers: Mickael Maison <mickael.maison@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Rajini Sivaram <rajinisivaram@googlemail.com>

78 min agoStreams Use case anchor (#4420)
manjuapu [Thu, 18 Jan 2018 17:59:32 +0000 (09:59 -0800)] 
Streams Use case anchor (#4420)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

81 min agoKAFKA-6456; JavaDoc clarification for Connect SourceTask#poll() (#4432)
Gunnar Morling [Thu, 18 Jan 2018 17:56:22 +0000 (18:56 +0100)] 
KAFKA-6456; JavaDoc clarification for Connect SourceTask#poll() (#4432)

Making clear that implementations of poll() shouldn't block indefinitely in order to allow the task instance to transition to PAUSED state.

Reviewers:  Randall Hauch <rhauch@gmail.com>, Jason Gustafson <jason@confluent.io>

114 min agoMINOR: Fix error message in KafkaConfig validation (#4417)
Andras Beni [Thu, 18 Jan 2018 17:23:23 +0000 (18:23 +0100)] 
MINOR: Fix error message in KafkaConfig validation (#4417)

2 hours agoMINOR: Fix typo in KafkaConsumer javadoc (#4422)
Magnus Reftel [Thu, 18 Jan 2018 17:11:20 +0000 (18:11 +0100)] 
MINOR: Fix typo in KafkaConsumer javadoc (#4422)

2 hours agoimprove internal topic integration test (#4437)
Guozhang Wang [Thu, 18 Jan 2018 16:57:42 +0000 (08:57 -0800)] 
improve internal topic integration test (#4437)

Reviewers: Damian Guy <damian.guy@gmail.com>

2 hours agoMINOR: Need to get a new transformer for each get() call. can't share'em (#4435)
dan norwood [Thu, 18 Jan 2018 16:45:22 +0000 (08:45 -0800)] 
MINOR: Need to get a new transformer for each get() call. can't share'em (#4435)

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Matthias J. Sax <mjsax@apache.org>

21 hours agoKAFKA-6302: Improve AdmintClient JavaDocs
Matthias J. Sax [Wed, 17 Jan 2018 22:01:43 +0000 (14:01 -0800)] 
KAFKA-6302: Improve AdmintClient JavaDocs

Reviewers: Colin P. McCabe <cmccabe@confluent.io>, Ismael Juma <ismael@juma.me.uk>

Closes #4332

23 hours agoKAFKA-6382: Make ProducerConfig and ConsumerConfig constructors public (#4341)
Matthias J. Sax [Wed, 17 Jan 2018 19:25:50 +0000 (11:25 -0800)] 
KAFKA-6382: Make ProducerConfig and ConsumerConfig constructors public (#4341)

* KIP-234
* update constructors to accept Properties and Map<String,Object>
* use ProducerConfig to access BATCH_SIZE default value in Streams

42 hours agoKAFKA-6328: Sort node groups considering global stores in InternalTopologyBuilder...
RichardYuSTUG [Wed, 17 Jan 2018 00:44:44 +0000 (16:44 -0800)] 
KAFKA-6328: Sort node groups considering global stores in InternalTopologyBuilder#makeNodeGroups

Author: RichardYuSTUG <yohan.richard.yu2@gmail.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #4339 from ConcurrencyPractitioner/kafka-6238

Minor edits on description

44 hours agoMINOR: Improve Join integration test coverage, PART II
Guozhang Wang [Tue, 16 Jan 2018 23:13:30 +0000 (15:13 -0800)] 
MINOR: Improve Join integration test coverage, PART II

1. Replaced KStreamKTableJoinIntegrationTest with the abstract based StreamTableJoinIntegrationTest. Added details on per-step verifications.
2. Minor renaming on GlobalKTableIntegrationTest.

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Damian Guy <damian.guy@gmail.com>, Bill Bejeck <bill@confluent.io>

Closes #4419 from guozhangwang/KMinor-join-integration-tests-II

2 days agoKAFKA-4218: Enable access to key in ValueTransformer and ValueMapper
Jeyhun Karimov [Tue, 16 Jan 2018 18:47:29 +0000 (10:47 -0800)] 
KAFKA-4218: Enable access to key in ValueTransformer and ValueMapper

This PR is the partial implementation for KIP-149. As the discussion for this KIP is still ongoing, I made a PR on the "safe" portions of the KIP (so that it can be included in the next release) which are 1) `ValueMapperWithKey`, 2) `ValueTransformerWithKeySupplier`, and 3) `ValueTransformerWithKey`.

Author: Jeyhun Karimov <je.karimov@gmail.com>

Reviewers: Damian Guy <damian.guy@gmail.com>, Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4309 from jeyhunkarimov/KIP-149_hope_last

2 days agoMINOR: Fix typo (#4426)
Joel Hamill [Tue, 16 Jan 2018 18:30:11 +0000 (10:30 -0800)] 
MINOR: Fix typo (#4426)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

2 days agoKAFKA-4541; Support for delegation token mechanism
Manikumar Reddy [Tue, 16 Jan 2018 17:50:31 +0000 (09:50 -0800)] 
KAFKA-4541; Support for delegation token mechanism

- Add capability to create delegation token
- Add authentication based on delegation token.
- Add capability to renew/expire delegation tokens.
- Add units tests and integration tests

Author: Manikumar Reddy <manikumar.reddy@gmail.com>

Reviewers: Jun Rao <junrao@gmail.com>

Closes #3616 from omkreddy/KAFKA-4541

3 days agoKAFKA-5890; records.lag should use tags for topic and partition rather than using...
cmolter [Mon, 15 Jan 2018 00:18:39 +0000 (16:18 -0800)] 
KAFKA-5890; records.lag should use tags for topic and partition rather than using metric name.

This is the implementation of KIP-225.
It marks the previous metrics as deprecated in the documentation and adds new metrics using tags.

Testing verifies that both the new and the old metric report the same value.

Author: cmolter <cmolter@apple.com>

Reviewers: Jiangjie (Becket) Qin <becket.qin@gmail.com>

Closes #4362 from lahabana/kafka-5890

5 days ago4385
RichardYuSTUG [Sat, 13 Jan 2018 01:27:52 +0000 (17:27 -0800)] 
4385

Author: RichardYuSTUG <yohan.richard.yu2@gmail.com>
Author: Prasanna Gautam <prasannagautam@gmail.com>
Author: Dong Lin <lindong28@gmail.com>
Author: Vahid Hashemian <vahidhashemian@us.ibm.com>
Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4385 from ConcurrencyPractitioner/doc-kafka-4499

5 days agoMINOR: Improve Join integration test coverage, PART I
Guozhang Wang [Fri, 12 Jan 2018 23:40:59 +0000 (15:40 -0800)] 
MINOR: Improve Join integration test coverage, PART I

0. Rename `JoinIntegrationTest` to `StreamStreamJoinIntegrationTest`, which is only for KStream-KStream joins.
1. Extract the `AbstractJoinIntegrationTest` which is going to be used for all the join integration test classes, parameterized with and without caching.
2. Merge `KStreamRepartitionJoinTest.java` into `StreamStreamJoinIntegrationTest.java` with augmented stream-stream join.
3. Add `TableTableJoinIntegrationTest` with detailed per-step expected results and removed `KTableKTableJoinIntegrationTest`.

Findings of the integration test:

1. Confirmed KAFKA-4309 with caching turned on.
2. Found bug KAFKA-6398.
3. Found bug KAFKA-6443.
4. Found a bug that in CachingKeyValueStore, we would flush before putting the record into the underlying store, when the store is going to be used in the downstream processors with flushing it would result in incorrect results, fixed the issue along with this PR.
5. Consider a new optimization described in KAFKA-6286.

Future works including stream-table joins will be in other PRs.

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Damian Guy <damian.guy@gmail.com>, Bill Bejeck <bill@confluent.io>

Closes #4331 from guozhangwang/KMinor-join-integration-tests

6 days agoKAFKA-6250: Use existing Kafka Connect internal topics without requiring ACL (#4247)
Gavrie Philipson [Thu, 11 Jan 2018 23:52:50 +0000 (01:52 +0200)] 
KAFKA-6250: Use existing Kafka Connect internal topics without requiring ACL (#4247)

When using Kafka Connect with a cluster that doesn't allow the user to
create topics (due to ACL configuration), Connect fails when trying to
create its internal topics even if these topics already exist. This is
incorrect behavior according to the documentation, which mentions that
R/W access should be enough.

This happens specifically when using Aiven Kafka, which does not permit
creation of topics via the Kafka Admin Client API.

The patch ignores the returned error, similar to the behavior for older
brokers that don't support the API.

7 days agoKAFKA-6265: Add #queryableStoreName() to GlobalKTable
Richard Yu [Thu, 11 Jan 2018 17:54:02 +0000 (09:54 -0800)] 
KAFKA-6265: Add #queryableStoreName() to GlobalKTable

A spinoff of original pull request #4340 for resolving conflicts.

Author: RichardYuSTUG <yohan.richard.yu2@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4413 from ConcurrencyPractitioner/kafka-6265-2

7 days agoKAFKA-6383: Complete shut down for streams threads that have not started
Rohan Desai [Thu, 11 Jan 2018 00:24:43 +0000 (16:24 -0800)] 
KAFKA-6383: Complete shut down for streams threads that have not started

*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*

*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*

Author: Rohan Desai <desai.p.rohan@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4382 from rodesai/KAFKA-6383

7 days agoMerge pull request #3567 from bobrik/unclean-docs-clarification
Jun Rao [Wed, 10 Jan 2018 21:32:16 +0000 (13:32 -0800)] 
Merge pull request #3567 from bobrik/unclean-docs-clarification

KAFKA-4711: fix docs onunclean.leader.election.enable default

8 days agoKAFKA-6398: fix KTable.filter that does not include its parent's queryable storename
Guozhang Wang [Wed, 10 Jan 2018 19:14:20 +0000 (11:14 -0800)] 
KAFKA-6398: fix KTable.filter that does not include its parent's queryable storename

1. Include the parent's queryable store name in KTable.filter if this operator is not materialized.
2. Augment InternalTopologyBuilder checking on null processor / store names from the enum.
3. Unit test.

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Damian Guy <damian.guy@gmail.com>, Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>

Closes #4384 from guozhangwang/K6398-topology-builder-exception

8 days agoKAFKA-6412 Improve synchronization in CachingKeyValueStore methods
tedyu [Wed, 10 Jan 2018 10:21:51 +0000 (10:21 +0000)] 
KAFKA-6412 Improve synchronization in CachingKeyValueStore methods

Currently CachingKeyValueStore methods are synchronized at method level.

It seems we can use read lock for getter and write lock for put / delete methods.

For getInternal(), if the underlying thread is streamThread, the getInternal() may trigger eviction. This can be handled by obtaining write lock at the beginning of the method for streamThread.

The jmh patch is attached to JIRA:
https://issues.apache.org/jira/secure/attachment/12905140/6412-jmh.v1.txt

Author: tedyu <yuzhihong@gmail.com>

Reviewers: Damian Guy <damian.guy@gmail.com>, Bill Bejeck <bill@confluent.io>

Closes #4372 from tedyu/6412

8 days agoMINOR: Menu updates and navigation (#4405)
Joel Hamill [Wed, 10 Jan 2018 00:28:49 +0000 (16:28 -0800)] 
MINOR: Menu updates and navigation (#4405)

* Menu updates and navigation

Reviewers: Guozhang Wang <wangguoz@gmail.com>

8 days agoKAFKA-4247; Prevent CLASSPATH from beginning with a single colon
Chris Egerton [Tue, 9 Jan 2018 22:49:18 +0000 (14:49 -0800)] 
KAFKA-4247; Prevent CLASSPATH from beginning with a single colon

Different fix for problem addressed by https://github.com/apache/kafka/pull/1953. Should prevent the CLASSPATH environment variable from being prefixed by a single colon before the JVM is invoked in the run-class script, which will then prevent the current working directory from being unintentionally included in the classpath when using the Reflections library.

If the current working directory should still be included in the classpath, it just needs to be explicitly specified either with its fully-qualified pathname or as a single dot (".").

Author: Chris Egerton <chrise@confluent.io>

Reviewers: Randall Hauch <rhauch@gmail.com>, Jun Rao <junrao@gmail.com>

Closes #4406 from C0urante/fix-colon-prefixed-classpath

9 days agoKAFKA-6096: Add multi-threaded tests for group coordinator, txn manager (#4122)
Rajini Sivaram [Tue, 9 Jan 2018 00:15:35 +0000 (00:15 +0000)] 
KAFKA-6096: Add multi-threaded tests for group coordinator, txn manager (#4122)

Reviewers: Jason Gustafson <jason@confluent.io>

9 days agoKAFKA-6422 Mirror maker will throw null pointer exception when the message value...
lisa2lisa [Mon, 8 Jan 2018 22:59:16 +0000 (23:59 +0100)] 
KAFKA-6422 Mirror maker will throw null pointer exception when the message value is null (#4387)

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, James Cheng <jylcheng@yahoo.com>, Jason Gustafson <jason@confluent.io>

9 days agoUpdating docs for streams app reset tool (#4401)
Jorge Quilcate Otoya [Mon, 8 Jan 2018 20:02:29 +0000 (21:02 +0100)] 
Updating docs for streams app reset tool (#4401)

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

9 days agoKAFKA-6363: Use MockAdminClient for any unit tests that depend on AdminClient (#4371)
Filipe Agapito [Mon, 8 Jan 2018 19:58:56 +0000 (19:58 +0000)] 
KAFKA-6363: Use MockAdminClient for any unit tests that depend on AdminClient (#4371)

* Implement MockAdminClient.deleteTopics
* Use MockAdminClient instead of MockKafkaAdminClientEnv in StreamsResetterTest
* Rename MockKafkaAdminClientEnv to AdminClientUnitTestEnv
* Use MockAdminClient instead of MockKafkaAdminClientEnv in TopicAdminTest
* Rename KafkaAdminClient to AdminClientUnitTestEnv in KafkaAdminClientTest.java
* Migrate StreamThreadTest to MockAdminClient
* Fix style errors
* Address review comments
* Fix MockAdminClient call

Reviewers: Matthias J. Sax <matthias@confluent.io>, Konstantine Karantasis <konstantine@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

9 days agoMINOR: Fix menu ordering in streams docs
Joel Hamill [Mon, 8 Jan 2018 19:43:38 +0000 (11:43 -0800)] 
MINOR: Fix menu ordering in streams docs

https://issues.apache.org/jira/browse/KAFKA-6419

Related: https://github.com/apache/kafka-site/pull/115

Author: Joel Hamill <joel-hamill@users.noreply.github.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #4390 from joel-hamill/fix-streams-menu-order

10 days agoMINOR: Improve exception messages when state stores cannot be accessed. (#4383)
Ewen Cheslack-Postava [Mon, 8 Jan 2018 19:14:54 +0000 (11:14 -0800)] 
MINOR: Improve exception messages when state stores cannot be accessed. (#4383)

10 days agoMINOR: fix image 404s in streams doc
Joel Hamill [Mon, 8 Jan 2018 17:38:58 +0000 (09:38 -0800)] 
MINOR: fix image 404s in streams doc

Fixes image 404s in streams docs (e.g. https://kafka.apache.org/documentation/streams/developer-guide/interactive-queries.html).

Related https://github.com/apache/kafka-site/pull/114

Author: Joel Hamill <joel-hamill@users.noreply.github.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #4389 from joel-hamill/fix-404-streams

10 days agoMINOR: Catch JsonMappingException subclass (#3821)
Romain Hardouin [Fri, 8 Sep 2017 22:21:00 +0000 (00:21 +0200)] 
MINOR: Catch JsonMappingException subclass (#3821)

Handle InvalidTypeIdException as NOT_IMPLEMENTED and add unit tests for all exceptions.

Reviewers: Colin P. Mccabe <cmccabe@confluent.io>, Rajini Sivaram <rajinisivaram@googlemail.com>

10 days agoMINOR: Add documentation for KAFKA-6086 (ProductionExceptionHandler) (#4395)
Matt Farmer [Mon, 8 Jan 2018 11:33:23 +0000 (06:33 -0500)] 
MINOR: Add documentation for KAFKA-6086 (ProductionExceptionHandler) (#4395)

* Update streams documentation to describe production exception handler

* Add a mention of the ProductionExceptionHandler in the upgrade guide

12 days agoKAFKA-6252; Close the metric group to clean up any existing metrics
Arjun Satish [Sat, 6 Jan 2018 05:23:39 +0000 (21:23 -0800)] 
KAFKA-6252; Close the metric group to clean up any existing metrics

We are closing the metricGroups created in a Worker, Source task and Sink task before populating them with new metrics. This helps in cases where an Exception is thrown when previously created groups were not cleaned up correctly.

Signed-off-by: Arjun Satish <arjunconfluent.io>
Author: Arjun Satish <arjun@confluent.io>

Reviewers: Randall Hauch <rhauch@gmail.com>, Konstantine Karantasis <konstantine@confluent.io>, Ewen Cheslack-Postava <ewen@confluent.io>

Closes #4397 from wicknicks/KAFKA-6252

12 days agoKAFKA-4335: Add batch.size to FileStreamSource connector to prevent OOM
Study [Sat, 6 Jan 2018 02:31:53 +0000 (18:31 -0800)] 
KAFKA-4335: Add batch.size to FileStreamSource connector to prevent OOM

When the source file of `FileStreamSource` is a large file, `FileStreamSourceTask.poll()` will result in OOM. This pull request added `batch.size` parameter which can restrict the poll size.

*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*

*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*

Author: Study <ph.study@gmail.com>

Reviewers: Ewen Cheslack-Postava <ewen@confluent.io>

Closes #4356 from phstudy/KAFKA-4335

13 days agoKAFKA-6311: Expose Kafka cluster ID in Connect REST API (KIP-238) (#4314)
Ewen Cheslack-Postava [Fri, 5 Jan 2018 15:52:50 +0000 (07:52 -0800)] 
KAFKA-6311: Expose Kafka cluster ID in Connect REST API (KIP-238) (#4314)

Reviewers: Konstantine Karantasis <konstantine@confluent.io>, Randall Hauch <rhauch@gmail.com>, Jason Gustafson <jason@confluent.io>

2 weeks agoMINOR: Fix concurrency bug in MetadataCache and Metadata request when listeners incon...
Ismael Juma [Thu, 4 Jan 2018 18:16:04 +0000 (18:16 +0000)] 
MINOR: Fix concurrency bug in MetadataCache and Metadata request when listeners inconsistent (#4374)

- Add missing locking/volatile in MetadataCache.aliveEndPoint
- Fix topic metadata not to throw BrokerNotAvailableException
when listeners are inconsistent. Add test verifying the fix. As
part of this fix, renamed Broker methods to follow Map
convention where the `get` version returns `Option`.

Reviewers: Jason Gustafson <jason@confluent.io>

2 weeks agoKAFKA-6298; Added support for location aware logger which fixes log line numbers...
Narendra kumar [Thu, 4 Jan 2018 16:59:53 +0000 (17:59 +0100)] 
KAFKA-6298; Added support for location aware logger which fixes log line numbers (#4311)

LogContext to have two different implementations of Logger. One will be picked based on availability of LocationAwareLogger API.

Reviewers: Jason Gustafson <jason@confluent.io>

2 weeks agoMINOR: Fix zk client session state metric names and various async zk clean-ups
Ismael Juma [Wed, 3 Jan 2018 22:02:14 +0000 (14:02 -0800)] 
MINOR: Fix zk client session state metric names and various async zk clean-ups

- Fix zk session state and session change rate metric names: type
should be SessionExpireListener instead of KafkaHealthCheck. Test
verifying the fix was included.
- Handle missing controller in controlled shutdown in the same way as if
the broker is not registered (i.e. retry after backoff).
- Restructure BrokerInfo to reduce duplication. It now contains a
Broker instance and the JSON serde is done in BrokerIdZNode
since `Broker` does not contain all the fields.
- Remove dead code from `ZooKeeperClient.initialize` and remove
redundant `close` calls.
- Move ACL handling and persistent paths definition from ZkUtils to
ZkData (and call ZkData from ZkUtils).
- Remove ZooKeeperClientWrapper and ZooKeeperClientMetrics from
ZkUtils (avoids metrics clash if third party users create a ZkUtils
instance in the same process as the broker).
- Introduce factory method in KafkaZkClient that creates
ZooKeeperClient and remove metric name defaults from
ZooKeeperClient.
- Fix a few instances where ZooKeeperClient was not closed in tests.
- Update a few TestUtils methods to use KafkaZkClient instead of
ZkUtils.
- Add test verifying SessionState metric.
- Various clean-ups.

Testing: mostly relying on existing tests, but added a couple
of new tests as mentioned above.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Jun Rao <junrao@gmail.com>

Closes #4359 from ijuma/kafka-6320-kafka-health-zk-metrics-follow-up

2 weeks agoKAFKA-6415; Use WARN log level for Metadata in system test
Rajini Sivaram [Wed, 3 Jan 2018 19:17:53 +0000 (19:17 +0000)] 
KAFKA-6415; Use WARN log level for Metadata in system test

When a log entry is appended to a Kafka topic using `KafkaLog4jAppender`, the producer.send operation may block waiting for metadata. This can result in deadlocks in a couple of scenarios if a log entry from the producer network thread is also at a log level that results in the entry being appended to a Kafka topic.
1. Producer's network thread will attempt to send data to a Kafka topic and this is unsafe since producer.send may block waiting for metadata, causing a deadlock since the thread will not process the metadata request/response.
2. `KafkaLog4jAppender#append` is invoked while holding the lock of the logger. So the thread waiting for metadata in the initial send will be holding the logger lock. If the producer network thread has.a log entry that needs to be appended, it will attempt to acquire the logger lock and deadlock.

This is a temporary workaround to avoid deadlocks in system tests by setting log level to WARN for `Metadata` in `VerifiableLog4jAppender`. The fix has been verified using the system tests log4j_appender_test.py which started failing when the info-level log entry was introduced.

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Satish Duggana <satish.duggana@gmail.com>, tedyu <yuzhihong@gmail.com>

Closes #4375 from rajinisivaram/KAFKA-6415-log4jappender

2 weeks agoKAFKA-6256: fix flaky test KStreamKTableJoinIntegrationTest.shouldCountClicksPerRegio...
Matthias J. Sax [Wed, 3 Jan 2018 19:08:43 +0000 (11:08 -0800)] 
KAFKA-6256: fix flaky test KStreamKTableJoinIntegrationTest.shouldCountClicksPerRegionWithNonZeroByteCache

Increase commit interval to make it less likely that we flush the cache in-between.
To make it fool-proof, only compare the "final" result records if cache is enabled.

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4364 from mjsax/kafka-6256-flaky-kstream-ktable-join-with-caching-test

2 weeks agoRevert "KAFKA-6383: complete shutdown for CREATED StreamThreads (#4343)"
Guozhang Wang [Tue, 2 Jan 2018 22:21:42 +0000 (14:21 -0800)] 
Revert "KAFKA-6383: complete shutdown for CREATED StreamThreads (#4343)"

This reverts commit 47db063c310cf47e4c544196acab2abfe62977b0.

2 weeks agoMINOR: refactored code duplicates in several files (Streams project) (#4357)
Wladimir Schmidt [Tue, 2 Jan 2018 18:54:20 +0000 (19:54 +0100)] 
MINOR: refactored code duplicates in several files (Streams project) (#4357)

* Removed code duplicate from GlobalProcessorContextImpl and ProcessorContextImpl to parent class AbstractProcessorContext

* Exchanged concrete implementations with interfaces to make code more maintainable

* Refactored major code duplicates in InternalTopologyBuilder

* Formatted function parameters as per code review
Added final to code introduced in this PR

* Added missing finals to putNodeGroupName function
Rearranged parameters for resetTopicsPattern function

Reviewers: Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <wangguoz@gmail.com>, Bill Bejeck <bbejeck@gmail.com>

2 weeks agoKAFKA-6318: StreamsResetter should return non-zero return code on error (#4305)
Siva Santhalingam [Tue, 2 Jan 2018 18:45:57 +0000 (10:45 -0800)] 
KAFKA-6318: StreamsResetter should return non-zero return code on error (#4305)

Reviewers: Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <wangguoz@gmail.com>, Ted Yu <yuzhihong@gmail.com>, Bill Bejeck <bbejeck@gmail.com>

2 weeks agoKAFKA-6391 ensure topics are created with correct partitions BEFORE building the...
Clemens Valiente [Tue, 2 Jan 2018 18:22:28 +0000 (19:22 +0100)] 
KAFKA-6391 ensure topics are created with correct partitions BEFORE building theā€¦ (#4347)

* ensure topics are created with correct partitions BEFORE building the metadata for our stream tasks

* Added a test case. The test should fail with the old logic, because:
While stream-partition-assignor-test-KSTREAM-MAP-0000000001-repartition is created correctly with four partitions, the StreamPartitionAssignor will only assign three tasks to the topic. Test passes with the new logic.

Reviewers: Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <wangguoz@gmail.com>, Ted Yu <yuzhihong@gmail.com>

2 weeks agoFix wrong property mentioned in doc; Author: Praveen K Palaniswamy <yourspraveen...
Guozhang Wang [Tue, 2 Jan 2018 18:07:54 +0000 (10:07 -0800)] 
Fix wrong property mentioned in doc; Author: Praveen K Palaniswamy <yourspraveen@gmail.com>

2 weeks agoKAFKA-6269: KTable restore fails after rebalance (#4300) 4370/head
Bill Bejeck [Tue, 2 Jan 2018 17:42:40 +0000 (12:42 -0500)] 
KAFKA-6269: KTable restore fails after rebalance (#4300)

* Return offset of next record of records left after restore completed
* Changed check for restoring partition to remove the "+1" in the guard condition

Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <wangguoz@gmail.com>

2 weeks agoKAFKA-5368: Add test for skipped-records metric (#4365)
Matthias J. Sax [Tue, 2 Jan 2018 17:34:31 +0000 (09:34 -0800)] 
KAFKA-5368: Add test for skipped-records metric (#4365)

* KAFKA-5368: Add test for skipped-records metric

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

2 weeks agoKAFKA-6383: complete shutdown for CREATED StreamThreads (#4343)
Rohan [Tue, 2 Jan 2018 17:32:17 +0000 (09:32 -0800)] 
KAFKA-6383: complete shutdown for CREATED StreamThreads (#4343)

* KAFKA-6383: complete shutdown for CREATED StreamThreads

When transitioning StreamThread from CREATED to PENDING_SHUTDOWN
free up resources from the caller, rather than the stream thread,
since in this case the stream thread was never actually started.

Have StreamThread.setState return the old state. If the old state is
CREATED in StreamThread.shutdown() then start the thread so that it
can free the resources owned by the StreamThread.

Add a KafkaStreams test to verify that the producer gets closed even
if KafkaStreams was not started

Reviewers: Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <wangguoz@gmail.com>

2 weeks agoReplace Arrays.asList with Collections.singletonList where possible (#4368)
Koen De Groote [Mon, 1 Jan 2018 13:07:32 +0000 (14:07 +0100)] 
Replace Arrays.asList with Collections.singletonList where possible (#4368)

Reviewers: Ismael Juma <ismael@juma.me.uk>

2 weeks agoKAFKA-6307: Fix KafkaMbean leak in JmxReporter (#4307)
tedyu [Mon, 1 Jan 2018 10:16:07 +0000 (02:16 -0800)] 
KAFKA-6307: Fix KafkaMbean leak in JmxReporter (#4307)

We should remove the map entry from mbeans if it becomes
empty during metric removal.

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Satish Duggana <satish.duggana@gmail.com>, Ismael Juma <ismael@juma.me.uk>

2 weeks agoMINOR: Temporarily disable flaky Streams EOS system tests (#4355)
Matthias J. Sax [Fri, 29 Dec 2017 12:44:08 +0000 (04:44 -0800)] 
MINOR: Temporarily disable flaky Streams EOS system tests (#4355)

Reviewers: Bill Bejeck <bill@confluent.io>, Ismael Juma <ismael@juma.me.uk>

2 weeks agoMINOR: Use EnumMap/EnumSet if possible (#3919)
Koen De Groote [Fri, 29 Dec 2017 12:39:18 +0000 (13:39 +0100)] 
MINOR: Use EnumMap/EnumSet if possible (#3919)

They are more efficient than HashMap/HashSet.

Reviewers: Ismael Juma <ismael@juma.me.uk>

3 weeks agoMINOR: Fixed clusterId reference in Metadata.update (#4360)
Satish Duggana [Thu, 28 Dec 2017 18:21:43 +0000 (23:51 +0530)] 
MINOR: Fixed clusterId reference in Metadata.update (#4360)

Also fixed log message with respective error in KerberosLogin.login.

Reviewers: Jason Gustafson <jason@confluent.io>

3 weeks agoMINOR: Update test classes to use KafkaZkClient/AdminZkClient methods (#4353)
Manikumar Reddy O [Wed, 27 Dec 2017 16:30:02 +0000 (22:00 +0530)] 
MINOR: Update test classes to use KafkaZkClient/AdminZkClient methods (#4353)

3 weeks agoMINOR: Use TopicPartition in ConsumerGroupCommand instead of TopicAndPartition where...
Vahid Hashemian [Sat, 23 Dec 2017 02:37:36 +0000 (18:37 -0800)] 
MINOR: Use TopicPartition in ConsumerGroupCommand instead of TopicAndPartition where possible (#4333)

Reviewers: Jason Gustafson <jason@confluent.io>

3 weeks agoMINOR: Web docs for KIP-220
Guozhang Wang [Fri, 22 Dec 2017 19:53:04 +0000 (11:53 -0800)] 
MINOR: Web docs for KIP-220

1. added functions for KafkaStreams and KafkaClientSupplier.
2. added prefix for admin client in StreamsConfig.

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Jason Gustafson <jason@confluent.io>, Matthias J. Sax <matthias@confluent.io>

Closes #4338 from guozhangwang/K6150-doc-changes

3 weeks agoKAFKA-5772 Improve Util classes (#3370)
Matthias Wessendorf [Fri, 22 Dec 2017 19:41:21 +0000 (20:41 +0100)] 
KAFKA-5772 Improve Util classes (#3370)

Utils with static methods should not be instantiated, hence marking the classes `final` and adding a `private` constructor.

Reviewers: Randall Hauch <rhauch@gmail.com>, Jason Gustafson <jason@confluent.io>

3 weeks agoKAFKA-6390: Update ZooKeeper to 3.4.11, Gradle and other minor updates (#4345)
Ismael Juma [Fri, 22 Dec 2017 18:41:22 +0000 (18:41 +0000)] 
KAFKA-6390: Update ZooKeeper to 3.4.11, Gradle and other minor updates (#4345)

Updates:
- Gradle, gradle plugins and maven artifact updated
- Bug fix updates for ZooKeeper, Jackson, EasyMock and Snappy

Not updated:
- RocksDB as it often causes issues, so better done separately
- args4j as our test coverage is weak and the update was a
feature release

Also fixed scala-reflect version to match scala-library.

Release notes for ZooKeeper 3.4.11:
https://zookeeper.apache.org/doc/r3.4.11/releasenotes.html

A notable fix is improved handling of UnknownHostException:
https://issues.apache.org/jira/browse/ZOOKEEPER-2614

Manually tested that IntelliJ import and build still works.
Relying on existing test suite otherwise.

Reviewers: Jun Rao <junrao@gmail.com>

3 weeks agoKAFKA-6320: Move ZK metrics in KafkaHealthCheck to ZookeeperClient (#4351)
Jun Rao [Fri, 22 Dec 2017 10:28:57 +0000 (02:28 -0800)] 
KAFKA-6320: Move ZK metrics in KafkaHealthCheck to ZookeeperClient (#4351)

* Moved metrics in KafkaHealthCheck to ZookeeperClient.
* Converted remaining ZkUtils usage in KafkaServer to ZookeeperClient and removed ZkUtils from KafkaServer.
* Made the re-creation of ZooKeeper during ZK session expiration with infinite retries.
* Added unit tests for all new methods in KafkaZkClient.

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Ismael Juma <ismael@juma.me.uk>

3 weeks agoKAFKA-5895: Gradle 3.0+ is needed on the build (#3861)
Matthias Wessendorf [Fri, 22 Dec 2017 01:52:12 +0000 (02:52 +0100)] 
KAFKA-5895: Gradle 3.0+ is needed on the build (#3861)

Reviewers: Ismael Juma <ismael@juma.me.uk>

3 weeks agoMINOR: Use GitHub git repo for push by default (#4352)
Ismael Juma [Fri, 22 Dec 2017 01:19:13 +0000 (01:19 +0000)] 
MINOR: Use GitHub git repo for push by default (#4352)

4 weeks agoMINOR: Improve Streams Dev Guide content on web docs
Joel Hamill [Thu, 21 Dec 2017 19:15:54 +0000 (11:15 -0800)] 
MINOR: Improve Streams Dev Guide content on web docs

This PR migrates content from CP Streams Dev Guide.

Here is the top-level page:
![image](https://user-images.githubusercontent.com/11722533/33904945-df9cf804-df31-11e7-93aa-52385961522c.png)

Here is a child page:
![image](https://user-images.githubusercontent.com/11722533/33904976-f2eafabe-df31-11e7-918c-fbf95db0f76b.png)

See related: https://github.com/apache/kafka-site/pull/112

Author: Joel Hamill <joel-hamill@users.noreply.github.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #4252 from joel-hamill/20171122-migrate-cp-dev-guide

4 weeks agoMINOR: Fix race condition in Streams EOS system test
Guozhang Wang [Thu, 21 Dec 2017 02:44:36 +0000 (18:44 -0800)] 
MINOR: Fix race condition in Streams EOS system test

We should start the process only within the `with` block, otherwise the bytes parameter would cause a race condition that result in false alarms of system test failures.

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Ewen Cheslack-Postava <me@ewencp.org>

Closes #4348 from guozhangwang/KMinor-fix-eos-test

4 weeks agoKAFKA-6126: Remove unnecessary topics created check
Matthias J. Sax [Thu, 21 Dec 2017 02:02:33 +0000 (18:02 -0800)] 
KAFKA-6126: Remove unnecessary topics created check

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4322 from mjsax/kafka-6126-remove-topic-check-on-rebalance-2

4 weeks agoKAFKA-4263: Fix flaky test QueryableStateIntegrationTest.concurrentAccess
Matthias J. Sax [Wed, 20 Dec 2017 23:22:06 +0000 (15:22 -0800)] 
KAFKA-4263: Fix flaky test QueryableStateIntegrationTest.concurrentAccess

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Damian Guy <damian.guy@gmail.com>, Guozhang Wang <wangguoz@gmail.com>

Closes #4342 from mjsax/kafka-4263-concurrentAccess

4 weeks agoKAFKA-5849; Add process stop, round trip workload, partitioned test
Colin P. Mccabe [Wed, 20 Dec 2017 21:35:33 +0000 (21:35 +0000)] 
KAFKA-5849; Add process stop, round trip workload, partitioned test

* Implement process stop faults via SIGSTOP / SIGCONT

* Implement RoundTripWorkload, which both sends messages, and confirms that they are received at least once.

* Allow Trogdor tasks to block until other Trogdor tasks are complete.

* Add CreateTopicsWorker, which can be a building block for a lot of tests.

* Simplify how TaskSpec subclasses in ducktape serialize themselves to JSON.

* Implement some fault injection tests in round_trip_workload_test.py

Author: Colin P. Mccabe <cmccabe@confluent.io>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #4323 from cmccabe/KAFKA-5849

4 weeks agoKAFKA-5647; Use KafkaZkClient in ReassignPartitionsCommand and PreferredReplicaLeader...
Manikumar Reddy [Wed, 20 Dec 2017 20:19:36 +0000 (12:19 -0800)] 
KAFKA-5647; Use KafkaZkClient in ReassignPartitionsCommand and PreferredReplicaLeaderElectionCommand

*  Use KafkaZkClient in ReassignPartitionsCommand
*  Use KafkaZkClient in PreferredReplicaLeaderElectionCommand
*  Updated test classes to use new methods
*  All existing tests should pass

Author: Manikumar Reddy <manikumar.reddy@gmail.com>

Reviewers: Jun Rao <junrao@gmail.com>

Closes #4260 from omkreddy/KAFKA-5647-ADMINCOMMANDS

4 weeks agoKAFKA-5746; Document new broker metrics added for health checks
Rajini Sivaram [Wed, 20 Dec 2017 13:37:42 +0000 (13:37 +0000)] 
KAFKA-5746; Document new broker metrics added for health checks

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #4026 from rajinisivaram/MINOR-KIP-188-metrics-docs

4 weeks agoKAFKA-6331; Fix transient failure in AdminClientIntegrationTest.testAlterReplicaLogDirs
Dong Lin [Wed, 20 Dec 2017 12:01:22 +0000 (12:01 +0000)] 
KAFKA-6331; Fix transient failure in AdminClientIntegrationTest.testAlterReplicaLogDirs

Author: Dong Lin <lindong28@gmail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #4306 from lindong28/KAFKA-6331

4 weeks agoKAFKA-6150: KIP-204 part III; Change repartition topic segment size and ms
Guozhang Wang [Wed, 20 Dec 2017 00:05:42 +0000 (16:05 -0800)] 
KAFKA-6150: KIP-204 part III; Change repartition topic segment size and ms

1. Create default internal topic configs in StreamsConfig, especially for repartition topics change the segment size and time to smaller value.
2. Consolidate the default internal topic settings to InternalTopicManager and simplify InternalTopicConfig correspondingly.
3. Add an integration test for purging data.
4. MINOR: change TopologyBuilderException to IllegalStateException in StreamPartitionAssignor (part of https://issues.apache.org/jira/browse/KAFKA-5660).

Here are a few public facing APIs that get added:

1. AbstractConfig#originalsWithPrefix(String prefix, boolean strip): this for simplify the logic of passing admin and topic prefixed configs to consumer properties.
2. KafkaStreams constructor with Time object for convienent mocking in tests.

Will update KIP-204 accordingly if people re-votes these changes.

Author: Guozhang Wang <wangguoz@gmail.com>
Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>, Damian Guy <damian.guy@gmail.com>

Closes #4315 from guozhangwang/K6150-segment-size

4 weeks agoMINOR: Broker down for significant amt of time system test
Bill Bejeck [Tue, 19 Dec 2017 23:37:21 +0000 (15:37 -0800)] 
MINOR: Broker down for significant amt of time system test

System test where a broker is offline more than the configured timeouts.  In this case:
- Max poll interval set to 45 secs
- Retries set to 2
- Request timeout set to 15 seconds
- Max block ms set to 30 seconds

The broker was taken off-line for 70 seconds or more than double request timeout * num retries

[passing system test results](http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2017-12-11--001.1513034559--bbejeck--KSTREAMS_1179_broker_down_for_significant_amt_of_time--6ab4802/report.html)

Author: Bill Bejeck <bill@confluent.io>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4313 from bbejeck/KSTREAMS_1179_broker_down_for_significant_amt_of_time

4 weeks agoKAFKA-6317; Maven artifact for kafka should not depend on log4j
Ismael Juma [Tue, 19 Dec 2017 15:15:27 +0000 (15:15 +0000)] 
KAFKA-6317; Maven artifact for kafka should not depend on log4j

It should only depend on slf4j-api (like kafka-clients). The
release tarball still includes log4j and slf4j-log4j12.

Manually verified that there are no duplicate dependencies
in the release tarball and `./gradlew core:dependencies`
looks good.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #4297 from ijuma/kafka-6317-kafka-slf4j-api-only

4 weeks agoMINOR: Fix race condition in KafkaStreamsTest.shouldReturnThreadMetadata
Matthias J. Sax [Tue, 19 Dec 2017 01:39:25 +0000 (17:39 -0800)] 
MINOR: Fix race condition in KafkaStreamsTest.shouldReturnThreadMetadata

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Closes #4337 from mjsax/minor-fix-kafakstreamstest

4 weeks agoMINOR: Stabilize flaky smoke system tests before KIP-91
Matthias J. Sax [Tue, 19 Dec 2017 01:34:50 +0000 (17:34 -0800)] 
MINOR: Stabilize flaky smoke system tests before KIP-91

This is a workaround until KIP-91 is merged. We tried increasing the timeout multiple times already but tests are still flaky.

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Bill Bejeck <bill@confluent.io>, Apurva Mehta <apurva@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4329 from mjsax/hotfix-system-tests

4 weeks agoKAFKA-6258; SSLTransportLayer should keep reading from socket until either the buffer...
Dong Lin [Mon, 18 Dec 2017 19:15:09 +0000 (11:15 -0800)] 
KAFKA-6258; SSLTransportLayer should keep reading from socket until either the buffer is full or the socket has no more data

When consumer uses plaintext and there is remaining data in consumer's buffer, consumer.poll() will read all data available from the socket buffer to consumer buffer. However, if consumer uses ssl and there is remaining data, consumer.poll() may only read 16 KB (the size of SslTransportLayer.appReadBuffer) from socket buffer. This will reduce efficient of consumer.poll() by asking user to call more poll() to get the same amount of data.

Furthermore, we observe that for users who naively sleep a constant time after each consumer.poll(), some partition will lag behind after they switch from plaintext to ssl. Here is the explanation why this can happen.

Say there are 1 partition of 1MB/sec and 9 partition of 32KB/sec. Leaders of these partitions are all different and consumer is consuming these 10 partitions. Let's also assume that socket read buffer size is large enough and consume sleeps 1 sec between consumer.poll(). 1 sec is long enough for consumer to receive the FetchResponse back from broker.

When consumer uses plaintext, each consumer.poll() will read all data from the socket buffer and it means 1 MB data is read from each partition.

When consumer uses ssl, each consumer.poll() is likely to find that there is some data available in the memory. In this case consumer only reads 16 KB data from other sockets, particularly the socket for the broker with the large partition. Then the throughput of the large partition will be limited to 16KB/sec.

Arguably user should not sleep 1 sec if its consumer is lagging behind. But on Kafka dev side it is nice to keep the previous behavior and optimize consumer.poll() to read as much data from socket as possible.

Author: Dong Lin <lindong28@gmail.com>

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>, Jason Gustafson <jason@confluent.io>

Closes #4248 from lindong28/KAFKA-6258

4 weeks agoMINOR: improve JavaDocs for KafkaStreams, KafkaProducer, KafkaConsumer
Matthias J. Sax [Mon, 18 Dec 2017 19:10:03 +0000 (11:10 -0800)] 
MINOR: improve JavaDocs for KafkaStreams, KafkaProducer, KafkaConsumer

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Apurva Mehta <apurva@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #4335 from mjsax/minor-improve-KafkaStreams-javadocs

4 weeks agoKAFKA-5473; handle ZK session expiration properly when a new session can't be established
Prasanna Gautam [Fri, 15 Dec 2017 22:48:30 +0000 (14:48 -0800)] 
KAFKA-5473; handle ZK session expiration properly when a new session can't be established

(WIP: this commit isn't ready to be reviewed yet. I was checking the travis-ci build with the configuration changes in my account and opened the PR prematurely against trunk. I will make it consistent with Contribution guidelines once it's well tested.)

https://issues.apache.org/jira/browse/KAFKA-5473

Design:
`zookeeper.connection.retry.timeout.ms` => this determines how long to wait before triggering the shutdown. The default is 60000ms.

Currently the implementation only handles the `handleSessionEstablishmentError` by waiting for the sessionTimeout.

Author: Prasanna Gautam <prasannagautam@gmail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jun Rao <junrao@gmail.com>

Closes #3990 from prasincs/KAFKA-5473

4 weeks agoKAFKA-5526; Additional `--describe` views for ConsumerGroupCommand (KIP-175)
Vahid Hashemian [Fri, 15 Dec 2017 18:26:00 +0000 (10:26 -0800)] 
KAFKA-5526; Additional `--describe` views for ConsumerGroupCommand (KIP-175)

The `--describe` option of ConsumerGroupCommand is expanded, as proposed in [KIP-175](https://cwiki.apache.org/confluence/display/KAFKA/KIP-175%3A+Additional+%27--describe%27+views+for+ConsumerGroupCommand), to support:
* `--describe` or `--describe --offsets`: listing of current group offsets
* `--describe --members` or `--describe --members --verbose`: listing of group members
* `--describe --state`: group status

Example: With a single partition topic `test1` and a double partition topic `test2`, consumers `consumer1` and `consumer11` subscribed to `test`, consumers `consumer2` and `consumer22` and `consumer222` subscribed to `test2`, and all consumers belonging to group `test-group`, this is an output example of the new options above for `test-group`:

```
--describe, or --describe --offsets:

TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                     HOST            CLIENT-ID
test2           0          0               0               0               consumer2-bad9496d-0889-47ab-98ff-af17d9460382  /127.0.0.1      consumer2
test2           1          0               0               0               consumer22-c45e6ee2-0c7d-44a3-94a8-9627f63fb411 /127.0.0.1      consumer22
test1           0          0               0               0               consumer1-d51b0345-3194-4305-80db-81a68fa6c5bf  /127.0.0.1      consumer1
```

```
--describe --members

CONSUMER-ID                                      HOST            CLIENT-ID       #PARTITIONS
consumer2-bad9496d-0889-47ab-98ff-af17d9460382   /127.0.0.1      consumer2       1
consumer222-ed2108cd-d368-41f1-8514-5b72aa835bcc /127.0.0.1      consumer222     0
consumer11-dc8295d7-8f3f-4438-9b11-7270bab46760  /127.0.0.1      consumer11      0
consumer22-c45e6ee2-0c7d-44a3-94a8-9627f63fb411  /127.0.0.1      consumer22      1
consumer1-d51b0345-3194-4305-80db-81a68fa6c5bf   /127.0.0.1      consumer1       1
```

```
--describe --members --verbose

CONSUMER-ID                                      HOST            CLIENT-ID       #PARTITIONS     ASSIGNMENT
consumer2-bad9496d-0889-47ab-98ff-af17d9460382   /127.0.0.1      consumer2       1               test2(0)
consumer222-ed2108cd-d368-41f1-8514-5b72aa835bcc /127.0.0.1      consumer222     0               -
consumer11-dc8295d7-8f3f-4438-9b11-7270bab46760  /127.0.0.1      consumer11      0               -
consumer22-c45e6ee2-0c7d-44a3-94a8-9627f63fb411  /127.0.0.1      consumer22      1               test2(1)
consumer1-d51b0345-3194-4305-80db-81a68fa6c5bf   /127.0.0.1      consumer1       1               test1(0)
```

```
--describe --state

COORDINATOR (ID)         ASSIGNMENT-STRATEGY       STATE                #MEMBERS
localhost:9092 (0)       range                     Stable               5
```

Note that this PR also addresses the issue reported in [KAFKA-6158](https://issues.apache.org/jira/browse/KAFKA-6158) by dynamically setting the width of columns `TOPIC`, `CONSUMER-ID`, `HOST`, `CLIENT-ID` and `COORDINATOR (ID)`. This avoid truncation of column values when they go over the current fixed width of these columns.

The code has been restructured to better support testing of individual values and also the console output. Unit tests have been updated and extended to take advantage of this restructuring.

Author: Vahid Hashemian <vahidhashemian@us.ibm.com>

Reviewers: Jason Gustafson <jason@confluent.io>

Closes #4271 from vahidhashemian/KAFKA-5526

4 weeks agoKAFKA-6086: Provide for custom error handling when Kafka Streams fails to produce
Matt Farmer [Fri, 15 Dec 2017 12:53:17 +0000 (12:53 +0000)] 
KAFKA-6086: Provide for custom error handling when Kafka Streams fails to produce

This PR creates and implements the `ProductionExceptionHandler` as described in [KIP-210](https://cwiki.apache.org/confluence/display/KAFKA/KIP-210+-+Provide+for+custom+error+handling++when+Kafka+Streams+fails+to+produce).

I've additionally provided a default implementation preserving the existing behavior. I fixed various compile errors in the tests that resulted from my changing of method signatures, and added tests to cover the new behavior.

Author: Matt Farmer <mfarmer@rsglab.com>
Author: Matt Farmer <matt@frmr.me>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>, Damian Guy <damian.guy@gmail.com>

Closes #4165 from farmdawgnation/msf/kafka-6086

4 weeks agoKAFKA-6308; Connect Struct should use deepEquals/deepHashCode
Tobias Gies [Thu, 14 Dec 2017 23:26:20 +0000 (15:26 -0800)] 
KAFKA-6308; Connect Struct should use deepEquals/deepHashCode

This changes the Struct's equals and hashCode method to use Arrays#deepEquals and Arrays#deepHashCode, respectively. This resolves a problem where two structs with values of type byte[] would not be considered equal even though the byte arrays' contents are equal. By using deepEquals, the byte arrays' contents are compared instead of ther identity.

Since this changes the behavior of the equals method for byte array values, the behavior of hashCode must change alongside it to ensure the methods still fulfill the general contract of "equal objects must have equal hashCodes".

Test rationale:
All existing unit tests for equals were untouched and continue to work. A new test method was added to verify the behavior of equals and hashCode for Struct instances that contain a byte array value. I verify the reflixivity and transitivity of equals as well as the fact that equal Structs have equal hashCodes
and not-equal structs do not have equal hashCodes.

Author: Tobias Gies <tobias.gies@trivago.com>
Author: Tobias Gies <tobias@tobiasgies.de>

Reviewers: Randall Hauch <rhauch@gmail.com>, Jason Gustafson <jason@confluent.io>

Closes #4293 from tobiasgies/feature/kafka-6308-deepequals

4 weeks agoKAFKA-6102; Consolidate MockTime implementations between connect and clients
Colin P. Mccabe [Thu, 14 Dec 2017 22:47:15 +0000 (14:47 -0800)] 
KAFKA-6102; Consolidate MockTime implementations between connect and clients

Author: Colin P. Mccabe <cmccabe@confluent.io>

Reviewers: Randall Hauch <rhauch@gmail.com>, Jason Gustafson <jason@confluent.io>

Closes #4105 from cmccabe/KAFKA-6102

5 weeks agoKAFKA-6360: Clear RocksDB Segments when store is closed
Damian Guy [Thu, 14 Dec 2017 17:51:56 +0000 (09:51 -0800)] 
KAFKA-6360: Clear RocksDB Segments when store is closed

Now that we support re-initializing state stores, we need to clear the segments when the store is closed so that they can be re-opened.

Author: Damian Guy <damian.guy@gmail.com>

Reviewers: Bill Bejeck <bbejeck@gmail.com>, Guozhang Wang <wangguoz@gmail.com>, Ted Yu <yuzhihong@gmail.com>

Closes #4324 from dguy/kafka-6360

5 weeks agoKAFKA-6349: Fix concurrent modification exception in AbstractStateManager during...
Damian Guy [Wed, 13 Dec 2017 03:25:01 +0000 (19:25 -0800)] 
KAFKA-6349: Fix concurrent modification exception in AbstractStateManager during restore

Fixes a `ConcurrentModificationException` in`AbstractStateManager` that is triggered when a `StateStore` is re-initialized and there are multiple stores in the context.

Author: Damian Guy <damian.guy@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>, GuozhangWang <wangguoz@gmail.com>

Closes #4317 from dguy/kafka-6349

5 weeks agoMINOR: Fix warn logging in RecordCollectorImpl
Damian Guy [Wed, 13 Dec 2017 03:13:39 +0000 (19:13 -0800)] 
MINOR: Fix warn logging in RecordCollectorImpl

Fix warn log message in RecordCollectorImpl so it prints the exception message rather than `{}`

Author: Damian Guy <damian.guy@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>, GuozhangWang <wangguoz@gmail.com>

Closes #4318 from dguy/minor-logging-record-collector

5 weeks agoKAFKA-5764; Add toLowerCase support to sasl.kerberos.principal.to.local rule (KIP...
Manikumar Reddy [Tue, 12 Dec 2017 23:47:49 +0000 (15:47 -0800)] 
KAFKA-5764; Add toLowerCase support to sasl.kerberos.principal.to.local rule (KIP-203)

Author: Manikumar Reddy <manikumar.reddy@gmail.com>

Reviewers: Jason Gustafson <jason@confluent.io>

Closes #3800 from omkreddy/KAFKA-5764-REGEX

5 weeks agoKAFKA-5631; Use Jackson for serialising to JSON
umesh chaudhary [Tue, 12 Dec 2017 14:00:04 +0000 (16:00 +0200)] 
KAFKA-5631; Use Jackson for serialising to JSON

- Rename `encode` to `legacyEncodeAsString`, we
can remove this when we remove `ZkUtils`.
- Introduce `encodeAsString` that uses Jackson.
- Change `encodeAsBytes` to use Jackson.
- Avoid intermediate string when converting
Broker to json bytes.

The methods that use Jackson only support
Java collections unlike `legacyEncodeAsString`.

Tests were added `encodeAsString` and
`encodeAsBytes`.

Author: umesh chaudhary <umesh9794@gmail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #4259 from umesh9794/KAFKA-5631

5 weeks agoKAFKA-6319: Quote strings stored in JSON configs
Rajini Sivaram [Tue, 12 Dec 2017 09:05:29 +0000 (11:05 +0200)] 
KAFKA-6319: Quote strings stored in JSON configs

This is required for ACLs where SSL principals contain
special characters (e.g. comma) that are escaped using
backslash. The strings need to be quoted for JSON to
ensure that the JSON stored in ZK is valid.

Also converted `SslEndToEndAuthorizationTest` to use a
principal with special characters to ensure that this
path is tested.

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #4303 from rajinisivaram/KAFKA-6319

5 weeks agoKAFKA-6324; Change LogSegment.delete to deleteIfExists and harden log recovery
Ismael Juma [Tue, 12 Dec 2017 08:00:05 +0000 (10:00 +0200)] 
KAFKA-6324; Change LogSegment.delete to deleteIfExists and harden log recovery

- Rename `delete()` to `deleteIfExists()` in `LogSegment`, `AbstractIndex`
and `TxnIndex`. Throw exception in case of IO errors for more informative
errors and to make it less likely that errors are ignored, `boolean` is used
for the case where the file does not exist (like `Files.deleteIfExists()`).
- Fix an instance of delete while open (should fix KAFKA-6322 and
KAFKA-6075).
- `LogSegment.deleteIfExists` no longer throws an exception if any of
the files it tries to delete does not exist (fixes KAFKA-6194).
- Remove unnecessary `FileChannel.force(true)` when deleting file.
- Introduce `LogSegment.open()` and use it to improve encapsulation
and reduce duplication.
- Expand functionality of `LogSegment.onBecomeInactiveSegment()`
to reduce duplication and improve encapsulation.
- Use `AbstractIndex.deleteIfExists()` instead of deleting files manually.
- Improve logging when deleting swap files.
- Use CorruptIndexException instead of IllegalArgumentException.
- Simplify `LogCleaner.cleanSegments()` to reduce duplication and
improve encapsulation.
- A few other clean-ups in Log, LogSegment, etc.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Jun Rao <junrao@gmail.com>, Ted Yu <yuzhihong@gmail.com>

Closes #4040 from ijuma/kafka-5829-follow-up

5 weeks agoKAFKA-6334: minor typo fix in web docs
Guozhang Wang [Tue, 12 Dec 2017 02:55:13 +0000 (18:55 -0800)] 
KAFKA-6334: minor typo fix in web docs

5 weeks agoMINOR: Increase number of messages in replica verification tool test
Rajini Sivaram [Mon, 11 Dec 2017 19:12:55 +0000 (11:12 -0800)] 
MINOR: Increase number of messages in replica verification tool test

Increase the number of messages produced to make the test more reliable. The test failed in a recent build and also fails intermittently when run locally. Since the producer uses acks=0 and the test stops as soon as a lag is observed, the change shouldn't have a big impact on the time taken to run when lag is observed sooner.

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Jason Gustafson <jason@confluent.io>

Closes #4312 from rajinisivaram/MINOR-replicaverification-test

5 weeks agoKAFKA-6121: Restore and global consumer should not use auto.offset.reset
Matthias J. Sax [Mon, 11 Dec 2017 14:20:10 +0000 (14:20 +0000)] 
KAFKA-6121: Restore and global consumer should not use auto.offset.reset

- set auto.offset.reste to "none" for restore and global consumer
- handle InvalidOffsetException for restore and global consumer
- add corresponding tests
- some minor cleanup

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Damian Guy <damian.guy@gmail.com, Bill Bejeck <bill@confluent.io>, GuozhangWang <wangguoz@gmail.com>

Closes #4215 from mjsax/kafka-6121-restore-global-consumer-handle-reset

5 weeks agoKAFKA-6289; NetworkClient should not expose failed internal ApiVersions requests 3998/head
Jason Gustafson [Fri, 8 Dec 2017 10:54:31 +0000 (10:54 +0000)] 
KAFKA-6289; NetworkClient should not expose failed internal ApiVersions requests

The NetworkClient internally ApiVersion requests to each broker following connection establishment. If this request happens to fail (perhaps due to an incompatible broker), the NetworkClient includes the response in the result of poll(). Applications will generally not be expecting this response which may lead to failed assertions (or in the case of AdminClient, an obscure log message).

I've added test cases which await the ApiVersion request sent by NetworkClient to be in-flight, and then disconnect the connection and verify that the response is not included from poll().

Author: Jason Gustafson <jason@confluent.io>

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #4280 from hachikuji/KAFKA-6289

5 weeks agoKAFKA-4857: Replace StreamsKafkaClient with AdminClient in Kafka Streams
Matthias J. Sax [Fri, 8 Dec 2017 00:16:54 +0000 (16:16 -0800)] 
KAFKA-4857: Replace StreamsKafkaClient with AdminClient in Kafka Streams

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Bill Bejeck <bbejeck@gmail.com>, Guozhang Wang <wangguoz@gmail.com>

Closes #4242 from mjsax/kafka-4857-admit-client

6 weeks agoKAFKA-6313; Add SLF4J as direct dependency to Kafka core
Randall Hauch [Thu, 7 Dec 2017 08:36:47 +0000 (10:36 +0200)] 
KAFKA-6313; Add SLF4J as direct dependency to Kafka core

Recent changes are now directly using the SLF4J API, so we should have a direct dependency.

Author: Randall Hauch <rhauch@gmail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #4296 from rhauch/kafka-6313

6 weeks agoMINOR: Avoid intermediate strings when parsing/decoding ZK JSON
Ismael Juma [Thu, 7 Dec 2017 00:38:34 +0000 (02:38 +0200)] 
MINOR: Avoid intermediate strings when parsing/decoding ZK JSON

Also:
- Fix bug in result type of `createSequentialPersistentPath`
- Remove duplicated code from `ReplicationUtils`
- Move `propagateIsrChanges` from `ReplicationUtils` to `KafkaZkClient`
- Add tests
- Minor clean-ups

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Jun Rao <junrao@gmail.com>, Ted Yu <yuzhihong@gmail.com>, Manikumar Reddy <manikumar.reddy@gmail.com>

Closes #4261 from ijuma/zk-data-improvements

6 weeks agoKAFKA-6065; Latency metric for KafkaZkClient
Ismael Juma [Wed, 6 Dec 2017 23:13:40 +0000 (01:13 +0200)] 
KAFKA-6065; Latency metric for KafkaZkClient

Measures the latency of each request.

Updated existing `ZkUtils` test to use `KafkaZkClient`
instead.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Jun Rao <junrao@gmail.com>

Closes #4265 from ijuma/kafka-6065-async-zk-metrics

6 weeks agoKAFKA-6193; Only delete reassign_partitions znode after reassignment is complete
Ismael Juma [Wed, 6 Dec 2017 20:18:41 +0000 (22:18 +0200)] 
KAFKA-6193; Only delete reassign_partitions znode after reassignment is complete

- Ensure that `partitionsBeingReassigned` is fully populated before
`removePartitionFromReassignedPartitions` is invoked. This is
necessary to avoid premature deletion of the `reassign_partitions`
znode.
- Modify and add tests to verify the fixes.
- Add documentation.
- Use `info` log message if assignedReplicas == newReplicas and
remove control flow based on exceptions.
- General logging improvements.
- Simplify `initializePartitionAssignment` by relying on logic already
present in `maybeTriggerPartitionReassignment`.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Jun Rao <junrao@gmail.com>, Guozhang Wang <wangguoz@gmail.com>

Closes #4283 from ijuma/kafka-6193-flaky-shouldPerformMultipleReassignmentOperationsOverVariousTopics

6 weeks agoKAFKA-6300; SelectorTest may fail with ConcurrentModificationException
tedyu [Wed, 6 Dec 2017 19:43:29 +0000 (19:43 +0000)] 
KAFKA-6300; SelectorTest may fail with ConcurrentModificationException

Synchronization is added w.r.t. sockets ArrayList to avoid ConcurrentModificationException

Author: tedyu <yuzhihong@gmail.com>

Reviewers: Viktor Somogyi <viktorsomogyi@gmail.com>, Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #4299 from tedyu/trunk