Re: [DISCUSS] KIP-291: Have separate queues for control requests and data requests

[Lucas Wang]

Hi All,

I've updated the KIP by adding the dedicated endpoints for controller
connections,
and pinning threads for controller requests.
Also I've updated the title of this KIP. Please take a look and let me know
your feedback.

Thanks a lot for your time!
Lucas

On Tue, Jul 24, 2018 at 10:19 AM, Mayuresh Gharat <
gharatmayures...@gmail.com> wrote:

> Hi Lucas,
> I agree, if we want to go forward with a separate controller plane and data
> plane and completely isolate them, having a separate port for controller
> with a separate Acceptor and a Processor sounds ideal to me.
>
> Thanks,
>
> Mayuresh
>
>
> On Mon, Jul 23, 2018 at 11:04 PM Becket Qin  wrote:
>
> > Hi Lucas,
> >
> > Yes, I agree that a dedicated end to end control flow would be ideal.
> >
> > Thanks,
> >
> > Jiangjie (Becket) Qin
> >
> > On Tue, Jul 24, 2018 at 1:05 PM, Lucas Wang 
> wrote:
> >
> > > Thanks for the comment, Becket.
> > > So far, we've been trying to avoid making any request handler thread
> > > special.
> > > But if we were to follow that path in order to make the two planes more
> > > isolated,
> > > what do you think about also having a dedicated processor thread,
> > > and dedicated port for the controller?
> > >
> > > Today one processor thread can handle multiple connections, let's say
> 100
> > > connections
> > >
> > > represented by connection0, ... connection99, among which
> connection0-98
> > > are from clients, while connection99 is from
> > >
> > > the controller. Further let's say after one selector polling, there are
> > > incoming requests on all connections.
> > >
> > > When the request queue is full, (either the data request being full in
> > the
> > > two queue design, or
> > >
> > > the one single queue being full in the deque design), the processor
> > thread
> > > will be blocked first
> > >
> > > when trying to enqueue the data request from connection0, then possibly
> > > blocked for the data request
> > >
> > > from connection1, ... etc even though the controller request is ready
> to
> > be
> > > enqueued.
> > >
> > > To solve this problem, it seems we would need to have a separate port
> > > dedicated to
> > >
> > > the controller, a dedicated processor thread, a dedicated controller
> > > request queue,
> > >
> > > and pinning of one request handler thread for controller requests.
> > >
> > > Thanks,
> > > Lucas
> > >
> > >
> > > On Mon, Jul 23, 2018 at 6:00 PM, Becket Qin 
> > wrote:
> > >
> > > > Personally I am not fond of the dequeue approach simply because it is
> > > > against the basic idea of isolating the controller plane and data
> > plane.
> > > > With a single dequeue, theoretically speaking the controller requests
> > can
> > > > starve the clients requests. I would prefer the approach with a
> > separate
> > > > controller request queue and a dedicated controller request handler
> > > thread.
> > > >
> > > > Thanks,
> > > >
> > > > Jiangjie (Becket) Qin
> > > >
> > > > On Tue, Jul 24, 2018 at 8:16 AM, Lucas Wang 
> > > wrote:
> > > >
> > > > > Sure, I can summarize the usage of correlation id. But before I do
> > > that,
> > > > it
> > > > > seems
> > > > > the same out-of-order processing can also happen to Produce
> requests
> > > sent
> > > > > by producers,
> > > > > following the same example you described earlier.
> > > > > If that's the case, I think this probably deserves a separate doc
> and
> > > > > design independent of this KIP.
> > > > >
> > > > > Lucas
> > > > >
> > > > >
> > > > >
> > > > > On Mon, Jul 23, 2018 at 12:39 PM, Dong Lin 
> > > wrote:
> > > > >
> > > > > > Hey Lucas,
> > > > > >
> > > > > > Could you update the KIP if you are confident with the approach
> > which
> > > > > uses
> > > > > > correlation id? The idea around correlation id is kind of
> scattered
> > > > > across
> > > > > > multiple emails. It will be useful if other reviews can read the
> > KIP
> > > to
> > > > > > understand the latest proposal.
> > > > > >
> > > > > > Thanks,
> > > > > > Dong
> > > > > >
> > > > > > On Mon, Jul 23, 2018 at 12:32 PM, Mayuresh Gharat <
> > > > > > gharatmayures...@gmail.com> wrote:
> > > > > >
> > > > > > > I like the idea of the dequeue implementation by Lucas. This
> will
> > > > help
> > > > > us
> > > > > > > avoid additional queue for controller and additional configs in
> > > > Kafka.
> > > > > > >
> > > > > > > Thanks,
> > > > > > >
> > > > > > > Mayuresh
> > > > > > >
> > > > > > > On Sun, Jul 22, 2018 at 2:58 AM Becket Qin <
> becket@gmail.com
> > >
> > > > > wrote:
> > > > > > >
> > > > > > > > Hi Jun,
> > > > > > > >
> > > > > > > > The usage of correlation ID might still be useful to address
> > the
> > > > > cases
> > > > > > > > that the controller epoch and leader epoch check are not
> > > sufficient
> > > > > to
> > > > > > > > guarantee correct behavior. For example, if the controller
> > sends
> > > a
> > > > > > > > LeaderAndIsrRequest followed by a StopReplicaRequest, and the
> > > > broker
> > > > > > > > processes it in the reverse 

[jira] [Created] (KAFKA-7207) Make rate & total metrics documentation consistent

[Kevin Lu (JIRA)]

Kevin Lu created KAFKA-7207:
---

 Summary: Make rate & total metrics documentation consistent
 Key: KAFKA-7207
 URL: https://issues.apache.org/jira/browse/KAFKA-7207
 Project: Kafka
  Issue Type: Improvement
  Components: documentation
Reporter: Kevin Lu
Assignee: Kevin Lu


[KIP-187|https://cwiki-test.apache.org/confluence/display/KAFKA/KIP-187+-+Add+cumulative+count+metric+for+all+Kafka+rate+metrics]
 added an additional cumulative total metric for each existing rate metric.

In the [Monitoring|https://kafka.apache.org/documentation/#monitoring] section 
of the docs, there is a mix of some sections only listing out the rate metric, 
and some sections listing out both the total metric and the rate metric. 

We should make all the sections consistent and list out all of the total 
metrics.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [VOTE] KIP-338 Support to exclude the internal topics in kafka-topics.sh command

[Chia-Ping Tsai]

> The voting has passed. Dong Lin (binding) is a committer.

Thanks for the kind reminder. Let me correct the voting result (see below):

+3 (binding)
1) Jason Gustafson
2) Gwen Shapira
3) Dong Lin

+3 (non-binding)
1) Colin McCabe
2) Manikumar
3) Vahid S Hashemian

KIP-338 has passed. Thank you all.

--
Chia-Ping

On 2018/07/25 18:38:26, Kamal Chandraprakash  
wrote: 
> Chia-Ping Tsai,
> 
> The voting has passed. Dong Lin (binding) is a committer.
> 
> On Wed, Jul 25, 2018 at 8:24 AM Chia-Ping Tsai  wrote:
> 
> > Current voting result is shown below.
> >
> > +2 (binding)
> > 1) Jason Gustafson
> > 2) Gwen Shapira
> >
> > +4 (non-binding)
> > 1) Colin McCabe
> > 2) Manikumar
> > 3) Dong Lin
> > 4) Vahid S Hashemian
> >
> > Need one more binding vote please
> >
> > Cheers,
> > Chia-Ping
> >
> > On 2018/07/16 08:10:51, Chia-Ping Tsai  wrote:
> > > hi folks,
> > >
> > > The discussion[1] of KIP-338[2] did not get any objection for last 6
> > days so it is time to start the voting thread.
> > >
> > > Thanks for your time!
> > >
> > > [1]
> > https://lists.apache.org/thread.html/9bd4e61b73c901b51132ada49743b9b703d40b85fc4eeaa5c9099900@%3Cdev.kafka.apache.org%3E
> > >
> > > [2]
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-338+Support+to+exclude+the+internal+topics+in+kafka-topics.sh+command
> > >
> > > Cheers,
> > > chia-ping
> > >
> >
> 

Re: [Vote] KIP-321: Update TopologyDescription to better represent Source and Sink Nodes

[Ted Yu]

+1

On Wed, Jul 25, 2018 at 7:24 PM Nishanth Pradeep 
wrote:

> Hello,
>
> I'm calling a vote for KIP-321:
>
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-321%3A+Update+TopologyDescription+to+better+represent+Source+and+Sink+Nodes
>
> Best,
> Nishanth Pradeep
>

[Vote] KIP-321: Update TopologyDescription to better represent Source and Sink Nodes

[Nishanth Pradeep]

Hello,

I'm calling a vote for KIP-321:

https://cwiki.apache.org/confluence/display/KAFKA/KIP-321%3A+Update+TopologyDescription+to+better+represent+Source+and+Sink+Nodes

Best,
Nishanth Pradeep

Re: [DISCUSS] KIP-258: Allow to Store Record Timestamps in RocksDB

[Matthias J. Sax]

Hi,

KIP-268 (rebalance meatadata) is finished and included in AK 2.0
release. Thus, I want to pick up this KIP again to get the RocksDB
upgrade done for 2.1.

I updated the KIP accordingly and also have a "prove of concept" PR
ready (for "in place" upgrade only):
https://github.com/apache/kafka/pull/5422/

There a still open questions, but I want to collect early feedback on
the proposed interfaces we need for the store upgrade. Also note, that
the KIP now also aim to define a generic upgrade path from any store
format A to any other store format B. Adding timestamps is just a
special case.

I will continue to work on the PR and refine the KIP in the meantime, too.

Looking forward to your feedback.

-Matthias


On 3/14/18 11:14 PM, Matthias J. Sax wrote:
> After some more thoughts, I want to follow John's suggestion and split
> upgrading the rebalance metadata from the store upgrade.
> 
> I extracted the metadata upgrade into it's own KIP:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-268%3A+Simplify+Kafka+Streams+Rebalance+Metadata+Upgrade
> 
> I'll update this KIP accordingly shortly. I also want to consider to
> make the store format upgrade more flexible/generic. Atm, the KIP is too
> much tailored to the DSL IMHO and does not encounter PAPI users that we
> should not force to upgrade the stores. I need to figure out the details
> and follow up later.
> 
> Please give feedback for the new KIP-268 on the corresponding discussion
> thread.
> 
> @James: unfortunately, for upgrading to 1.2 I couldn't figure out a way
> for a single rolling bounce upgrade. But KIP-268 proposes a fix for
> future upgrades. Please share your thoughts.
> 
> Thanks for all your feedback!
> 
> -Matthias
> 
> On 3/12/18 11:56 PM, Matthias J. Sax wrote:
>> @John: yes, we would throw if configs are missing (it's an
>> implementation details IMHO and thus I did not include it in the KIP)
>>
>> @Guozhang:
>>
>> 1) I understand know what you mean. We can certainly, allow all values
>> "0.10.0.x", "0.10.1.x", "0.10.2.x", ... "1.1.x" for `upgrade.from`
>> parameter. I had a similar though once but decided to collapse them into
>> one -- will update the KIP accordingly.
>>
>> 2) The idea to avoid any config would be, to always send both request.
>> If we add a config to eventually disable the old request, we don't gain
>> anything with this approach. The question is really, if we are willing
>> to pay this overhead from 1.2 on -- note, it would be limited to 2
>> versions and not grow further in future releases. More details in (3)
>>
>> 3) Yes, this approach subsumes (2) for later releases and allows us to
>> stay with 2 "assignment strategies" we need to register, as the new
>> assignment strategy will allow to "upgrade itself" via "version
>> probing". Thus, (2) would only be a workaround to avoid a config if
>> people upgrade from pre-1.2 releases.
>>
>> Thus, I don't think we need to register new "assignment strategies" and
>> send empty subscriptions for older version.
>>
>> 4) I agree that this is a tricky thing to get right with a single
>> rebalance. I share the concern that an application might never catch up
>> and thus the hot standby will never be ready.
>>
>> Maybe it's better to go with 2 rebalances for store upgrades. If we do
>> this, we also don't need to go with (2) and can get (3) in place for
>> future upgrades. I also think that changes to the metadata are more
>> likely and thus allowing for single rolling bounce for this case is more
>> important anyway. If we assume that store upgrade a rare, it might be ok
>> to sacrifice two rolling bounced for this case. It was just an idea I
>> wanted to share (even if I see the issues).
>>
>>
>> -Matthias
>>
>>
>>
>> On 3/12/18 11:45 AM, Guozhang Wang wrote:
>>> Hello Matthias, thanks for your replies.
>>>
>>>
>>> 1) About the config names: actually I was trying to not expose
>>> implementation details :) My main concern was that in your proposal the
>>> values need to cover the span of all the versions that are actually using
>>> the same version, i.e. "0.10.1.x-1.1.x". So if I (as a user) am upgrading
>>> from any versions within this range I need to remember to use the value
>>> "0.10.1.x-1.1.x" than just specifying my old version. In my suggestion I
>>> was trying to argue the benefit of just letting users to specify the actual
>>> Kafka version she's trying to upgrade from, than specifying a range of
>>> versions. I was not suggesting to use "v1, v2, v3" etc as the values, but
>>> still using Kafka versions like broker's `internal.version` config. But if
>>> you were suggesting the same thing, i.e. by "0.10.1.x-1.1.x" you meant to
>>> say users can just specify "0.10.1" or "0.10.2" or "0.11.0" or "1.1" which
>>> are all recognizable config values then I think we are actually on the same
>>> page.
>>>
>>> 2) About the "multi-assignment" idea: yes it would increase the network
>>> footprint, but not the message size, IF I'm not mis-understanding your idea
>>> 

Build failed in Jenkins: kafka-2.0-jdk8 #91

[Apache Jenkins Server]

See 


Changes:

[jason] MINOR: Add thread dumps if broker node cannot be stopped (#5373)

--
[...truncated 2.48 MB...]

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

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

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

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

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

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

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftInner[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftInner[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftOuter[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftOuter[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftLeft[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftLeft[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterLeft[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterLeft[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInner[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInner[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuter[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuter[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeft[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeft[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInnerInner[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInnerInner[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInnerOuter[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInnerOuter[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInnerLeft[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInnerLeft[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterInner[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterInner[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterOuter[caching enabled = true] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterOuter[caching enabled = true] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftInner[caching enabled = false] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftInner[caching enabled = false] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftOuter[caching enabled = false] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftOuter[caching enabled = false] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftLeft[caching enabled = false] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testLeftLeft[caching enabled = false] PASSED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterLeft[caching enabled = false] STARTED

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testOuterLeft[caching enabled = false] FAILED
java.lang.AssertionError: Condition not met within timeout 12. Topics 
not deleted after 12 milli seconds.

org.apache.kafka.streams.integration.TableTableJoinIntegrationTest > 
testInner[caching enabled = false] STARTED

[jira] [Resolved] (KAFKA-6014) new consumer mirror maker halts after committing offsets to a deleted topic

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-6014.

Resolution: Resolved

We changed the behavior of the consumer in KAFKA-6829 so that unknown topic 
errors are retriable in order to be consistent with how the error is handled 
everywhere else and because there are cases where this error may be based on 
stale metadata. That should resolve this issue, but someone can reopen if 
necessary.

> new consumer mirror maker halts after committing offsets to a deleted topic
> ---
>
> Key: KAFKA-6014
> URL: https://issues.apache.org/jira/browse/KAFKA-6014
> Project: Kafka
>  Issue Type: Bug
>Reporter: Onur Karaman
>Assignee: Jason Gustafson
>Priority: Major
>
> New consumer throws an unexpected KafkaException when trying to commit to a 
> topic that has been deleted. MirrorMaker.commitOffsets doesn't attempt to 
> catch the KafkaException and just kills the process. We didn't see this in 
> the old consumer because old consumer just silently drops failed offset 
> commits.
> I ran a quick experiment locally to prove the behavior. The experiment:
> 1. start up a single broker
> 2. create a single-partition topic t
> 3. create a new consumer that consumes topic t
> 4. make the consumer commit every few seconds
> 5. delete topic t
> 6. expect: KafkaException that kills the process.
> Here's my script:
> {code}
> package org.apache.kafka.clients.consumer;
> import org.apache.kafka.common.TopicPartition;
> import java.util.Collections;
> import java.util.List;
> import java.util.Properties;
> public class OffsetCommitTopicDeletionTest {
> public static void main(String[] args) throws InterruptedException {
> Properties props = new Properties();
> props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
> "localhost:9090");
> props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g");
> props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
> "org.apache.kafka.common.serialization.ByteArrayDeserializer");
> props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
> "org.apache.kafka.common.serialization.ByteArrayDeserializer");
> props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
> KafkaConsumer kafkaConsumer = new 
> KafkaConsumer<>(props);
> TopicPartition partition = new TopicPartition("t", 0);
> List partitions = 
> Collections.singletonList(partition);
> kafkaConsumer.assign(partitions);
> while (true) {
> kafkaConsumer.commitSync(Collections.singletonMap(partition, new 
> OffsetAndMetadata(0, "")));
> Thread.sleep(1000);
> }
> }
> }
> {code}
> Here are the other commands:
> {code}
> > rm -rf /tmp/zookeeper/ /tmp/kafka-logs* logs*
> > ./gradlew clean jar
> > ./bin/zookeeper-server-start.sh config/zookeeper.properties
> > export LOG_DIR=logs0 && ./bin/kafka-server-start.sh 
> > config/server0.properties
> > ./bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic t 
> > --partitions 1 --replication-factor 1
> > ./bin/kafka-run-class.sh 
> > org.apache.kafka.clients.consumer.OffsetCommitTopicDeletionTest
> > ./bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic t
> {code}
> Here is the output:
> {code}
> [2017-10-04 20:00:14,451] ERROR [Consumer clientId=consumer-1, groupId=g] 
> Offset commit failed on partition t-0 at offset 0: This server does not host 
> this topic-partition. 
> (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
> Exception in thread "main" org.apache.kafka.common.KafkaException: Partition 
> t-0 may not exist or user may not have Describe access to topic
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:789)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:734)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:808)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:788)
>   at 
> org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204)
>   at 
> org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167)
>   at 
> org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:506)
>   at 
> 

[jira] [Created] (KAFKA-7206) Enable batching in FindCoordinator

[Yishun Guan (JIRA)]

Yishun Guan created KAFKA-7206:
--

 Summary: Enable batching in FindCoordinator
 Key: KAFKA-7206
 URL: https://issues.apache.org/jira/browse/KAFKA-7206
 Project: Kafka
  Issue Type: Improvement
  Components: admin
Reporter: Yishun Guan


To quote [~guozhang]

"The proposal is that, we extend FindCoordinatorRequest to have multiple 
consumer ids: today each FindCoordinatorRequest only contains a single consumer 
id, so in our scenario we need to send N request for N consumer groups still. 
If we can request for coordinators in a single request, then the workflow could 
be simplified to:
 # send a single FindCoordinatorRequest to a broker asking for coordinators of 
all consumer groups.
1.a) note that the response may still succeed in finding some coordinators 
while error on others, and we need to handle them on that granularity (see 
below).
 # and then for the collected coordinator, group them by coordinator id and 
send one request per coordinator destination.

Note that this change would require the version to be bumped up, to 
{{FIND_COORDINATOR_REQUEST_V3}}for such protocol changes, also the RESPONSE 
version should be bumped up in order to include multiple coordinators."



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [DISCUSS] KIP-289: Improve the default group id behavior in KafkaConsumer

[Jason Gustafson]

Hi Vahid,

I was thinking we'd only use the old API version if we had to. That is,
only if the user has explicitly configured "" as the group.id. Otherwise,
we'd just use the new one. Another option is to just drop support in the
client for the empty group id, but usually we allow a deprecation period
for changes like this.

-Jason

On Wed, Jul 25, 2018 at 12:49 PM, Vahid S Hashemian <
vahidhashem...@us.ibm.com> wrote:

> Hi Jason,
>
> Thanks for additional clarification.
>
> So the next version of the OffsetCommit API will return an
> INVALID_GROUP_ID error for empty group ids; but on the client side we call
> the older version of the client until the next major release.
> The table below should summarize this.
>
>   +-+
>   | Client (group.id="")|
>   +-+
>   | pre-2.1 |   2.1  |   3.0|
> +-+---+-++--+
> | | V5 (cur.) | works   | works  | works|
> + API +---+-++--+
> | | V6| N/A | N/A (calls V5/warning) | INVALID_GROUP_ID |
> +-+---+-++--+
>
> Assumptions:
> * 2.1: The target release version for this KIP
> * 3.0: The next major release
>
> Please advise if you see an issue; otherwise, I'll update the KIP
> accordingly.
>
> Thanks!
> --Vahid
>
>
>
>
> From:   Jason Gustafson 
> To: dev 
> Date:   07/25/2018 12:08 AM
> Subject:***UNCHECKED*** Re: [DISCUSS] KIP-289: Improve the default
> group idbehavior in KafkaConsumer
>
>
>
> Hey Vahid,
>
> Sorry for the confusion. I think we all agree that going forward, we
> shouldn't support the empty group id, so the question is just around
> compatibility. I think we have to bump the OffsetCommit API version so
> that
> old clients which are unknowingly depending on the default empty group id
> will continue to work with new brokers. For new versions of the client, we
> can either drop support for the empty group id immediately or we can give
> users a grace period. I was thinking we would do the latter. We can change
> the default group.id, but in the case that a user has explicitly
> configured
> the empty group, then we can just use an old version of the OffsetCommit
> API which still supports it. In a future release, we can drop this support
> and only use the latest OffsetCommit version. Does that make sense?
>
> Thanks,
> Jason
>
>
> On Tue, Jul 24, 2018 at 12:36 PM, Vahid S Hashemian <
> vahidhashem...@us.ibm.com> wrote:
>
> > Hi Jason,
> >
> > Thanks for clarifying.
> >
> > So if we are going to continue supporting the empty group id as before
> > (with only an addition of a deprecation warning), and disable
> > enable.auto.commit for the new default (null) group id on the client
> side,
> > do we really need to bump up the OffsetCommit version?
> >
> > You mentioned "If an explicit empty string is configured for the group
> id,
> > then maybe we keep the current behavior for compatibility" which makes
> > sense to me, but I find it in conflict with your earlier suggestion "we
> > just need to bump the OffsetCommit request API and only accept the
> offset
> > commit for older versions.". Maybe I'm missing something?
> >
> > Thanks!
> > --Vahid
> >
> >
> >
> >
> > From:   Jason Gustafson 
> > To: dev 
> > Date:   07/23/2018 10:52 PM
> > Subject:Re: [DISCUSS] KIP-289: Improve the default group id
> > behavior in KafkaConsumer
> >
> >
> >
> > Hey Vahid,
> >
> > Thanks for the updates. Just to clarify, I was suggesting that we
> disable
> > enable.auto.commit only if no explicit group.id is configured. If an
> > explicit empty string is configured for the group id, then maybe we keep
> > the current behavior for compatibility. We can log a warning mentioning
> > the
> > deprecation and we can use the old version of the OffsetCommit API that
> > allows the empty group id. In a later release, we can drop this support
> in
> > the client. Does that seem reasonable?
> >
> > By the way, instead of using the new ILLEGAL_OFFSET_COMMIT error code,
> > couldn't we use INVALID_GROUP_ID?
> >
> > Thanks,
> > Jason
> >
> >
> >
> > On Mon, Jul 23, 2018 at 5:14 PM, Stanislav Kozlovski
> >  > > wrote:
> >
> > > Hey Vahid,
> > >
> > > No I don't see an issue with it. I believe it to be the best approach.
> > >
> > > Best,
> > > Stanisav
> > >
> > > On Mon, Jul 23, 2018 at 12:41 PM Vahid S Hashemian <
> > > vahidhashem...@us.ibm.com> wrote:
> > >
> > > > Hi Stanislav,
> > > >
> > > > Thanks for the feedback.
> > > > Do you see an issue with using `null` as the default group id (as
> > > > addressed by Jason in his response)?
> > > > This default group id would not support offset commits and consumers
> > 

Re: [DISCUSS] KIP-320: Allow fetchers to detect and handle log truncation

[Jason Gustafson]

Hi All,

I have made some updates to the KIP. As many of you know, a side project of
mine has been specifying the Kafka replication protocol in TLA. You can
check out the code here if you are interested:
https://github.com/hachikuji/kafka-specification. In addition to uncovering
a couple unknown bugs in the replication protocol (e.g.
https://issues.apache.org/jira/browse/KAFKA-7128), this has helped me
validate the behavior in this KIP. In fact, the original version I proposed
had a weakness. I initially suggested letting the leader validate the
expected epoch at the fetch offset. This made sense for the consumer in the
handling of unclean leader election, but it was not strong enough to
protect the follower in all cases. In order to make advancement of the high
watermark safe, for example, the leader actually needs to be sure that
every follower in the ISR matches its own epoch.

I attempted to fix this problem by treating the epoch in the fetch request
slightly differently for consumers and followers. For consumers, it would
be the expected epoch of the record at the fetch offset, and the leader
would raise a LOG_TRUNCATION error if the expectation failed. For
followers, it would be the current epoch and the leader would require that
it match its own epoch. This was unsatisfying both because of the
inconsistency in behavior and because the consumer was left with the weaker
fencing that we already knew was insufficient for the replicas. Ultimately
I decided that we should make the behavior consistent and that meant that
the consumer needed to act more like a following replica. Instead of
checking for truncation while fetching, the consumer should check for
truncation after leader changes. After checking for truncation, the
consumer can then use the current epoch when fetching and get the stronger
protection that it provides. What this means is that the Metadata API must
include the current leader epoch. Given the problems we have had around
stale metadata and how challenging they have been to debug, I'm convinced
that this is a good idea in any case and it resolves the inconsistent
behavior in the Fetch API. The downside is that there will be some
additional overhead upon leader changes, but I don't think it is a major
concern since leader changes are rare and the OffsetForLeaderEpoch request
is cheap.

This approach leaves the door open for some interesting follow up
improvements. For example, now that we have the leader epoch in the
Metadata request, we can implement similar fencing for the Produce API. And
now that the consumer can reason about truncation, we could consider having
a configuration to expose records beyond the high watermark. This would let
users trade lower end-to-end latency for weaker durability semantics. It is
sort of like having an acks=0 option for the consumer. Neither of these
options are included in this KIP, I am just mentioning them as potential
work for the future.

Finally, based on the discussion in this thread, I have added the
seekToCommitted API for the consumer. Please take a look and let me know
what you think.

Thanks,
Jason

On Fri, Jul 20, 2018 at 2:34 PM, Guozhang Wang  wrote:

> Hi Jason,
>
> The proposed API seems reasonable to me too. Could you please also update
> the wiki page (
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 320%3A+Allow+fetchers+to+detect+and+handle+log+truncation)
> with a section say "workflow" on how the proposed API will be co-used with
> others to:
>
> 1. consumer callers handling a LogTruncationException.
> 2. consumer internals for handling a retriable UnknownLeaderEpochException.
>
>
> Guozhang
>
>
> On Tue, Jul 17, 2018 at 10:23 AM, Anna Povzner  wrote:
>
> > Hi Jason,
> >
> >
> > I also like your proposal and agree that KafkaConsumer#seekToCommitted()
> > is
> > more intuitive as a way to initialize both consumer's position and its
> > fetch state.
> >
> >
> > My understanding that KafkaConsumer#seekToCommitted() is purely for
> > clients
> > who store their offsets externally, right? And we are still going to
> > add KafkaConsumer#findOffsets()
> > in this KIP as we discussed, so that the client can handle
> > LogTruncationException?
> >
> >
> > Thanks,
> >
> > Anna
> >
> >
> > On Thu, Jul 12, 2018 at 3:57 PM Dong Lin  wrote:
> >
> > > Hey Jason,
> > >
> > > It is a great summary. The solution sounds good. I might have minor
> > > comments regarding the method name. But we can discuss that minor
> points
> > > later after we reach consensus on the high level API.
> > >
> > > Thanks,
> > > Dong
> > >
> > >
> > > On Thu, Jul 12, 2018 at 11:41 AM, Jason Gustafson 
> > > wrote:
> > >
> > > > Hey Anna and Dong,
> > > >
> > > > Thanks a lot for the great discussion. I've been hanging back a bit
> > > because
> > > > honestly the best option hasn't seemed clear. I agree with Anna's
> > general
> > > > observation that there is a distinction between the position of the
> > > > consumer and its fetch state up to that position. 

[jira] [Created] (KAFKA-7205) KafkaConsumer / KafkaProducer should allow Reconfiguration of SSL Configuration

[Magnus Jungsbluth (JIRA)]

Magnus Jungsbluth created KAFKA-7205:


 Summary: KafkaConsumer / KafkaProducer should allow 
Reconfiguration of SSL Configuration
 Key: KAFKA-7205
 URL: https://issues.apache.org/jira/browse/KAFKA-7205
 Project: Kafka
  Issue Type: New Feature
  Components: clients
Affects Versions: 1.1.1
Reporter: Magnus Jungsbluth


Since Kafka 1.1 it is possible to reconfigure KeyStores on the broker side of 
things. 

When being serious about short lived keys, the client side should also support 
reconfiguring consumers and producers.

What I would propose is to implement {{Reconfigurable}}  on {{KafkaConsumer}} 
and {{KafkaProducer}}. The implementation has to pass the calls to 
NetworkClient which passes them on to Selector until they finally reach 
{{SslFactory}} which already implements {{Reconfigurable}}.

This seems pretty straightforward unless I am missing something important.  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [DISCUSS] KIP-289: Improve the default group id behavior in KafkaConsumer

[Vahid S Hashemian]

Hi Jason,

Thanks for additional clarification.

So the next version of the OffsetCommit API will return an 
INVALID_GROUP_ID error for empty group ids; but on the client side we call 
the older version of the client until the next major release.
The table below should summarize this.

  +-+
  | Client (group.id="")|
  +-+
  | pre-2.1 |   2.1  |   3.0|
+-+---+-++--+
| | V5 (cur.) | works   | works  | works|
+ API +---+-++--+
| | V6| N/A | N/A (calls V5/warning) | INVALID_GROUP_ID |
+-+---+-++--+

Assumptions:
* 2.1: The target release version for this KIP
* 3.0: The next major release

Please advise if you see an issue; otherwise, I'll update the KIP 
accordingly.

Thanks!
--Vahid




From:   Jason Gustafson 
To: dev 
Date:   07/25/2018 12:08 AM
Subject:***UNCHECKED*** Re: [DISCUSS] KIP-289: Improve the default 
group idbehavior in KafkaConsumer



Hey Vahid,

Sorry for the confusion. I think we all agree that going forward, we
shouldn't support the empty group id, so the question is just around
compatibility. I think we have to bump the OffsetCommit API version so 
that
old clients which are unknowingly depending on the default empty group id
will continue to work with new brokers. For new versions of the client, we
can either drop support for the empty group id immediately or we can give
users a grace period. I was thinking we would do the latter. We can change
the default group.id, but in the case that a user has explicitly 
configured
the empty group, then we can just use an old version of the OffsetCommit
API which still supports it. In a future release, we can drop this support
and only use the latest OffsetCommit version. Does that make sense?

Thanks,
Jason


On Tue, Jul 24, 2018 at 12:36 PM, Vahid S Hashemian <
vahidhashem...@us.ibm.com> wrote:

> Hi Jason,
>
> Thanks for clarifying.
>
> So if we are going to continue supporting the empty group id as before
> (with only an addition of a deprecation warning), and disable
> enable.auto.commit for the new default (null) group id on the client 
side,
> do we really need to bump up the OffsetCommit version?
>
> You mentioned "If an explicit empty string is configured for the group 
id,
> then maybe we keep the current behavior for compatibility" which makes
> sense to me, but I find it in conflict with your earlier suggestion "we
> just need to bump the OffsetCommit request API and only accept the 
offset
> commit for older versions.". Maybe I'm missing something?
>
> Thanks!
> --Vahid
>
>
>
>
> From:   Jason Gustafson 
> To: dev 
> Date:   07/23/2018 10:52 PM
> Subject:Re: [DISCUSS] KIP-289: Improve the default group id
> behavior in KafkaConsumer
>
>
>
> Hey Vahid,
>
> Thanks for the updates. Just to clarify, I was suggesting that we 
disable
> enable.auto.commit only if no explicit group.id is configured. If an
> explicit empty string is configured for the group id, then maybe we keep
> the current behavior for compatibility. We can log a warning mentioning
> the
> deprecation and we can use the old version of the OffsetCommit API that
> allows the empty group id. In a later release, we can drop this support 
in
> the client. Does that seem reasonable?
>
> By the way, instead of using the new ILLEGAL_OFFSET_COMMIT error code,
> couldn't we use INVALID_GROUP_ID?
>
> Thanks,
> Jason
>
>
>
> On Mon, Jul 23, 2018 at 5:14 PM, Stanislav Kozlovski
>  > wrote:
>
> > Hey Vahid,
> >
> > No I don't see an issue with it. I believe it to be the best approach.
> >
> > Best,
> > Stanisav
> >
> > On Mon, Jul 23, 2018 at 12:41 PM Vahid S Hashemian <
> > vahidhashem...@us.ibm.com> wrote:
> >
> > > Hi Stanislav,
> > >
> > > Thanks for the feedback.
> > > Do you see an issue with using `null` as the default group id (as
> > > addressed by Jason in his response)?
> > > This default group id would not support offset commits and consumers
> > would
> > > use `auto.offset.reset` config when there is no current offset.
> > >
> > > Thanks.
> > > --Vahid
> > >
> > >
> > >
> > >
> > > From:   Stanislav Kozlovski 
> > > To: dev@kafka.apache.org
> > > Date:   07/20/2018 11:09 AM
> > > Subject:Re: [DISCUSS] KIP-289: Improve the default group id
> > > behavior in KafkaConsumer
> > >
> > >
> > >
> > > I agree with Jason's notion that
> > > >  implicit use of the empty group.id to commit offsets is more 
likely
> > to
> > > be causing users unexpected problems than actually providing a 
useful
> > > capability.
> > > I was initially confused that this is the behavior when 
investigating
> a
> > 

[jira] [Created] (KAFKA-7204) MockConsumer poll clears all records in poll(), including records for subscriptions that are paused.

[Subodh Bhattacharjya (JIRA)]

Subodh Bhattacharjya created KAFKA-7204:
---

 Summary: MockConsumer poll clears all records in poll(), including 
records for subscriptions that are paused.
 Key: KAFKA-7204
 URL: https://issues.apache.org/jira/browse/KAFKA-7204
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Affects Versions: 1.1.0
Reporter: Subodh Bhattacharjya






--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [DISCUSS] KIP-346 - Limit blast radius of log compaction failure

[Dhruvil Shah]

For the cleaner thread specifically, I do not think respawning will help at
all because we are more than likely to run into the same issue again which
would end up crashing the cleaner. Retrying makes sense for transient
errors or when you believe some part of the system could have healed
itself, both of which I think are not true for the log cleaner.

On Wed, Jul 25, 2018 at 11:08 AM Ron Dagostino  wrote:

> << infinite loop which consumes resources and fires off continuous log
> messages.
> Hi Colin.  In case it could be relevant, one way to mitigate this effect is
> to implement a backoff mechanism (if a second respawn is to occur then wait
> for 1 minute before doing it; then if a third respawn is to occur wait for
> 2 minutes before doing it; then 4 minutes, 8 minutes, etc. up to some max
> wait time).
>
> I have no opinion on whether respawn is appropriate or not in this context,
> but a mitigation like the increasing backoff described above may be
> relevant in weighing the pros and cons.
>
> Ron
>
> On Wed, Jul 25, 2018 at 1:26 PM Colin McCabe  wrote:
>
> > On Mon, Jul 23, 2018, at 23:20, James Cheng wrote:
> > > Hi Stanislav! Thanks for this KIP!
> > >
> > > I agree that it would be good if the LogCleaner were more tolerant of
> > > errors. Currently, as you said, once it dies, it stays dead.
> > >
> > > Things are better now than they used to be. We have the metric
> > >   kafka.log:type=LogCleanerManager,name=time-since-last-run-ms
> > > which we can use to tell us if the threads are dead. And as of 1.1.0,
> we
> > > have KIP-226, which allows you to restart the log cleaner thread,
> > > without requiring a broker restart.
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration
> > > <
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration
> >
> >
> > > I've only read about this, I haven't personally tried it.
> >
> > Thanks for pointing this out, James!  Stanislav, we should probably add a
> > sentence or two mentioning the KIP-226 changes somewhere in the KIP.
> Maybe
> > in the intro section?
> >
> > I think it's clear that requiring the users to manually restart the log
> > cleaner is not a very good solution.  But it's good to know that it's a
> > possibility on some older releases.
> >
> > >
> > > Some comments:
> > > * I like the idea of having the log cleaner continue to clean as many
> > > partitions as it can, skipping over the problematic ones if possible.
> > >
> > > * If the log cleaner thread dies, I think it should automatically be
> > > revived. Your KIP attempts to do that by catching exceptions during
> > > execution, but I think we should go all the way and make sure that a
> new
> > > one gets created, if the thread ever dies.
> >
> > This is inconsistent with the way the rest of Kafka works.  We don't
> > automatically re-create other threads in the broker if they terminate.
> In
> > general, if there is a serious bug in the code, respawning threads is
> > likely to make things worse, by putting you in an infinite loop which
> > consumes resources and fires off continuous log messages.
> >
> > >
> > > * It might be worth trying to re-clean the uncleanable partitions. I've
> > > seen cases where an uncleanable partition later became cleanable. I
> > > unfortunately don't remember how that happened, but I remember being
> > > surprised when I discovered it. It might have been something like a
> > > follower was uncleanable but after a leader election happened, the log
> > > truncated and it was then cleanable again. I'm not sure.
> >
> > James, I disagree.  We had this behavior in the Hadoop Distributed File
> > System (HDFS) and it was a constant source of user problems.
> >
> > What would happen is disks would just go bad over time.  The DataNode
> > would notice this and take them offline.  But then, due to some
> > "optimistic" code, the DataNode would periodically try to re-add them to
> > the system.  Then one of two things would happen: the disk would just
> fail
> > immediately again, or it would appear to work and then fail after a short
> > amount of time.
> >
> > The way the disk failed was normally having an I/O request take a really
> > long time and time out.  So a bunch of request handler threads would
> > basically slam into a brick wall when they tried to access the bad disk,
> > slowing the DataNode to a crawl.  It was even worse in the second
> scenario,
> > if the disk appeared to work for a while, but then failed.  Any data that
> > had been written on that DataNode to that disk would be lost, and we
> would
> > need to re-replicate it.
> >
> > Disks aren't biological systems-- they don't heal over time.  Once
> they're
> > bad, they stay bad.  The log cleaner needs to be robust against cases
> where
> > the disk really is failing, and really is returning bad data or timing
> out.
> >
> > >
> > > * For your metrics, can you spell out the full metric in JMX-style
> > > 

Re: [DISCUSS] KIP-342 Add Customizable SASL extensions to OAuthBearer authentication

[Stanislav Kozlovski]

Hi Rajini,

I updated the KIP. Please check if the clarification is okay

On Wed, Jul 25, 2018 at 10:49 AM Rajini Sivaram 
wrote:

> Hi Stanislav,
>
> 1. Can you clarify the following line in the KIP in the 'Public Interfaces'
> section? When you are reading the KIP for the first time, it sounds like we
> adding a new Kafka config. But we are adding JAAS config options with a
> prefix that can be used with the default unsecured bearer tokens. We could
> include the example in this section or at least link to the example.
>
>- New config option for default, unsecured bearer tokens -
>`unsecuredLoginExtension_`.
>
>
> 2. Can you add the package for SaslExtensionsCallback class?
>
>
> On Tue, Jul 24, 2018 at 10:03 PM, Stanislav Kozlovski <
> stanis...@confluent.io> wrote:
>
> > Hi Ron,
> >
> > Thanks for the suggestions. I have applied them to the KIP.
> >
> > On Tue, Jul 24, 2018 at 1:39 PM Ron Dagostino  wrote:
> >
> > > Hi Stanislav.  The statement "New config option for
> > OAuthBearerLoginModule"
> > > is technically incorrect; it should be "New config option for default,
> > > unsecured bearer tokens" since that is what provides the functionality
> > (as
> > > opposed to the login module, which does not).  Also, please state that
> > > "auth" is not supported as a custom extension name with any
> > > SASL/OAUTHBEARER mechanism, including the unsecured one, since it is
> > > reserved by the spec for what is normally sent in the HTTP
> Authorization
> > > header an attempt to use it will result in a configuration exception.
> > >
> > > Finally, please also state that while the OAuthBearerLoginModule and
> the
> > > OAuthBearerSaslClient will be changed to request the extensions from
> its
> > > callback handler, for backwards compatibility it is not necessary for
> the
> > > callback handler to support SaslExtensionsCallback -- any
> > > UnsupportedCallbackException that is thrown will be ignored and no
> > > extensions will be added.
> > >
> > > Ron
> > >
> > > On Tue, Jul 24, 2018 at 11:20 AM Stanislav Kozlovski <
> > > stanis...@confluent.io>
> > > wrote:
> > >
> > > > Hey everybody,
> > > >
> > > > I have updated the KIP to reflect the latest changes as best as I
> > could.
> > > If
> > > > there aren't more suggestions, I intent to start the [VOTE] thread
> > > > tomorrow.
> > > >
> > > > Best,
> > > > Stanislav
> > > >
> > > > On Tue, Jul 24, 2018 at 6:34 AM Ron Dagostino 
> > wrote:
> > > >
> > > > > Hi Stanislav.  Could you update the KIP to reflect the latest
> > > definition
> > > > of
> > > > > SaslExtensions and confirm or correct the impact it has to the
> > > > > SCRAM-related classes?  I'm not sure if the currently-described
> > impact
> > > is
> > > > > still accurate.  Also, could you mention the changes to
> > > > > OAuthBearerUnsecuredLoginCallbackHandler in the text in addition to
> > > > giving
> > > > > the examples?  The examples show the new
> > > > > unsecuredLoginExtension_ feature, but that feature
> is
> > > not
> > > > > described anywhere prior to it appearing there.
> > > > >
> > > > > Ron
> > > > >
> > > > > On Mon, Jul 23, 2018 at 1:42 PM Ron Dagostino 
> > > wrote:
> > > > >
> > > > > > Hi Rajini.  I think a class is fine as long as we make sure the
> > > > semantics
> > > > > > of immutability are clear -- it would have to be a value class,
> and
> > > any
> > > > > > constructor that accepts a Map as input would have to copy that
> Map
> > > > > rather
> > > > > > than store it in a member variable.  Similarly, any Map that it
> > might
> > > > > > return would have to be unmodifiable.
> > > > > >
> > > > > > Ron
> > > > > >
> > > > > > On Mon, Jul 23, 2018 at 12:24 PM Rajini Sivaram <
> > > > rajinisiva...@gmail.com
> > > > > >
> > > > > > wrote:
> > > > > >
> > > > > >> Hi Ron, Stanislav,
> > > > > >>
> > > > > >> I agree with Stanislav that it would be better to leave
> > > > `SaslExtensions`
> > > > > >> as
> > > > > >> a class rather than make it an interface. We don''t really
> expect
> > > > users
> > > > > to
> > > > > >> extends this class, so it is convenient to have an
> implementation
> > > > since
> > > > > >> users need to create an instance. The class provided by the
> public
> > > API
> > > > > >> should be sufficient in the vast majority of the cases. Ron, do
> > you
> > > > > agree?
> > > > > >>
> > > > > >> On Mon, Jul 23, 2018 at 11:35 AM, Ron Dagostino <
> > rndg...@gmail.com>
> > > > > >> wrote:
> > > > > >>
> > > > > >> > Hi Stanislav.  See
> > > https://tools.ietf.org/html/rfc7628#section-3.1,
> > > > > and
> > > > > >> > that section refers to the core ABNF productions defined in
> > > > > >> > https://tools.ietf.org/html/rfc5234#appendix-B.
> > > > > >> >
> > > > > >> > Ron
> > > > > >> >
> > > > > >> > > On Jul 23, 2018, at 1:30 AM, Stanislav Kozlovski <
> > > > > >> stanis...@confluent.io>
> > > > > >> > wrote:
> > > > > >> > >
> > > > > >> > > Hey Ron and Rajini,
> > > > > >> > >
> > > > > >> > > Here are my thoughts:
> > > > > >> > 

Re: [VOTE] KIP-338 Support to exclude the internal topics in kafka-topics.sh command

[Kamal Chandraprakash]

Chia-Ping Tsai,

The voting has passed. Dong Lin (binding) is a committer.

On Wed, Jul 25, 2018 at 8:24 AM Chia-Ping Tsai  wrote:

> Current voting result is shown below.
>
> +2 (binding)
> 1) Jason Gustafson
> 2) Gwen Shapira
>
> +4 (non-binding)
> 1) Colin McCabe
> 2) Manikumar
> 3) Dong Lin
> 4) Vahid S Hashemian
>
> Need one more binding vote please
>
> Cheers,
> Chia-Ping
>
> On 2018/07/16 08:10:51, Chia-Ping Tsai  wrote:
> > hi folks,
> >
> > The discussion[1] of KIP-338[2] did not get any objection for last 6
> days so it is time to start the voting thread.
> >
> > Thanks for your time!
> >
> > [1]
> https://lists.apache.org/thread.html/9bd4e61b73c901b51132ada49743b9b703d40b85fc4eeaa5c9099900@%3Cdev.kafka.apache.org%3E
> >
> > [2]
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-338+Support+to+exclude+the+internal+topics+in+kafka-topics.sh+command
> >
> > Cheers,
> > chia-ping
> >
>

Re: [DISCUSS] KIP-346 - Limit blast radius of log compaction failure

[Ron Dagostino]

<< wrote:

> On Mon, Jul 23, 2018, at 23:20, James Cheng wrote:
> > Hi Stanislav! Thanks for this KIP!
> >
> > I agree that it would be good if the LogCleaner were more tolerant of
> > errors. Currently, as you said, once it dies, it stays dead.
> >
> > Things are better now than they used to be. We have the metric
> >   kafka.log:type=LogCleanerManager,name=time-since-last-run-ms
> > which we can use to tell us if the threads are dead. And as of 1.1.0, we
> > have KIP-226, which allows you to restart the log cleaner thread,
> > without requiring a broker restart.
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration
> > <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration>
>
> > I've only read about this, I haven't personally tried it.
>
> Thanks for pointing this out, James!  Stanislav, we should probably add a
> sentence or two mentioning the KIP-226 changes somewhere in the KIP.  Maybe
> in the intro section?
>
> I think it's clear that requiring the users to manually restart the log
> cleaner is not a very good solution.  But it's good to know that it's a
> possibility on some older releases.
>
> >
> > Some comments:
> > * I like the idea of having the log cleaner continue to clean as many
> > partitions as it can, skipping over the problematic ones if possible.
> >
> > * If the log cleaner thread dies, I think it should automatically be
> > revived. Your KIP attempts to do that by catching exceptions during
> > execution, but I think we should go all the way and make sure that a new
> > one gets created, if the thread ever dies.
>
> This is inconsistent with the way the rest of Kafka works.  We don't
> automatically re-create other threads in the broker if they terminate.  In
> general, if there is a serious bug in the code, respawning threads is
> likely to make things worse, by putting you in an infinite loop which
> consumes resources and fires off continuous log messages.
>
> >
> > * It might be worth trying to re-clean the uncleanable partitions. I've
> > seen cases where an uncleanable partition later became cleanable. I
> > unfortunately don't remember how that happened, but I remember being
> > surprised when I discovered it. It might have been something like a
> > follower was uncleanable but after a leader election happened, the log
> > truncated and it was then cleanable again. I'm not sure.
>
> James, I disagree.  We had this behavior in the Hadoop Distributed File
> System (HDFS) and it was a constant source of user problems.
>
> What would happen is disks would just go bad over time.  The DataNode
> would notice this and take them offline.  But then, due to some
> "optimistic" code, the DataNode would periodically try to re-add them to
> the system.  Then one of two things would happen: the disk would just fail
> immediately again, or it would appear to work and then fail after a short
> amount of time.
>
> The way the disk failed was normally having an I/O request take a really
> long time and time out.  So a bunch of request handler threads would
> basically slam into a brick wall when they tried to access the bad disk,
> slowing the DataNode to a crawl.  It was even worse in the second scenario,
> if the disk appeared to work for a while, but then failed.  Any data that
> had been written on that DataNode to that disk would be lost, and we would
> need to re-replicate it.
>
> Disks aren't biological systems-- they don't heal over time.  Once they're
> bad, they stay bad.  The log cleaner needs to be robust against cases where
> the disk really is failing, and really is returning bad data or timing out.
>
> >
> > * For your metrics, can you spell out the full metric in JMX-style
> > format, such as:
> >   kafka.log:type=LogCleanerManager,name=uncleanable-partitions-count
> >   value=4
> >
> > * For "uncleanable-partitions": topic-partition names can be very long.
> > I think the current max size is 210 characters (or maybe 240-ish?).
> > Having the "uncleanable-partitions" being a list could be very large
> > metric. Also, having the metric come out as a csv might be difficult to
> > work with for monitoring systems. If we *did* want the topic names to be
> > accessible, what do you think of having the
> >   kafka.log:type=LogCleanerManager,topic=topic1,partition=2
> > I'm not sure if LogCleanerManager is the right type, but my example was
> > that the topic and partition can be tags in the metric. That will allow
> > monitoring systems to more easily slice and dice the metric. I'm not
> > sure what the attribute for that metric would be. Maybe something like
> > "uncleaned bytes" for that topic-partition? Or time-since-last-clean? Or
> > maybe even just "Value=1".
>
> I haven't though about this that hard, but do we really need the
> uncleanable topic names to be accessible through a metric?  It seems like
> the admin should notice that uncleanable partitions are present, and 

Re: [DISCUSS] KIP-342 Add Customizable SASL extensions to OAuthBearer authentication

[Rajini Sivaram]

Hi Stanislav,

1. Can you clarify the following line in the KIP in the 'Public Interfaces'
section? When you are reading the KIP for the first time, it sounds like we
adding a new Kafka config. But we are adding JAAS config options with a
prefix that can be used with the default unsecured bearer tokens. We could
include the example in this section or at least link to the example.

   - New config option for default, unsecured bearer tokens -
   `unsecuredLoginExtension_`.


2. Can you add the package for SaslExtensionsCallback class?


On Tue, Jul 24, 2018 at 10:03 PM, Stanislav Kozlovski <
stanis...@confluent.io> wrote:

> Hi Ron,
>
> Thanks for the suggestions. I have applied them to the KIP.
>
> On Tue, Jul 24, 2018 at 1:39 PM Ron Dagostino  wrote:
>
> > Hi Stanislav.  The statement "New config option for
> OAuthBearerLoginModule"
> > is technically incorrect; it should be "New config option for default,
> > unsecured bearer tokens" since that is what provides the functionality
> (as
> > opposed to the login module, which does not).  Also, please state that
> > "auth" is not supported as a custom extension name with any
> > SASL/OAUTHBEARER mechanism, including the unsecured one, since it is
> > reserved by the spec for what is normally sent in the HTTP Authorization
> > header an attempt to use it will result in a configuration exception.
> >
> > Finally, please also state that while the OAuthBearerLoginModule and the
> > OAuthBearerSaslClient will be changed to request the extensions from its
> > callback handler, for backwards compatibility it is not necessary for the
> > callback handler to support SaslExtensionsCallback -- any
> > UnsupportedCallbackException that is thrown will be ignored and no
> > extensions will be added.
> >
> > Ron
> >
> > On Tue, Jul 24, 2018 at 11:20 AM Stanislav Kozlovski <
> > stanis...@confluent.io>
> > wrote:
> >
> > > Hey everybody,
> > >
> > > I have updated the KIP to reflect the latest changes as best as I
> could.
> > If
> > > there aren't more suggestions, I intent to start the [VOTE] thread
> > > tomorrow.
> > >
> > > Best,
> > > Stanislav
> > >
> > > On Tue, Jul 24, 2018 at 6:34 AM Ron Dagostino 
> wrote:
> > >
> > > > Hi Stanislav.  Could you update the KIP to reflect the latest
> > definition
> > > of
> > > > SaslExtensions and confirm or correct the impact it has to the
> > > > SCRAM-related classes?  I'm not sure if the currently-described
> impact
> > is
> > > > still accurate.  Also, could you mention the changes to
> > > > OAuthBearerUnsecuredLoginCallbackHandler in the text in addition to
> > > giving
> > > > the examples?  The examples show the new
> > > > unsecuredLoginExtension_ feature, but that feature is
> > not
> > > > described anywhere prior to it appearing there.
> > > >
> > > > Ron
> > > >
> > > > On Mon, Jul 23, 2018 at 1:42 PM Ron Dagostino 
> > wrote:
> > > >
> > > > > Hi Rajini.  I think a class is fine as long as we make sure the
> > > semantics
> > > > > of immutability are clear -- it would have to be a value class, and
> > any
> > > > > constructor that accepts a Map as input would have to copy that Map
> > > > rather
> > > > > than store it in a member variable.  Similarly, any Map that it
> might
> > > > > return would have to be unmodifiable.
> > > > >
> > > > > Ron
> > > > >
> > > > > On Mon, Jul 23, 2018 at 12:24 PM Rajini Sivaram <
> > > rajinisiva...@gmail.com
> > > > >
> > > > > wrote:
> > > > >
> > > > >> Hi Ron, Stanislav,
> > > > >>
> > > > >> I agree with Stanislav that it would be better to leave
> > > `SaslExtensions`
> > > > >> as
> > > > >> a class rather than make it an interface. We don''t really expect
> > > users
> > > > to
> > > > >> extends this class, so it is convenient to have an implementation
> > > since
> > > > >> users need to create an instance. The class provided by the public
> > API
> > > > >> should be sufficient in the vast majority of the cases. Ron, do
> you
> > > > agree?
> > > > >>
> > > > >> On Mon, Jul 23, 2018 at 11:35 AM, Ron Dagostino <
> rndg...@gmail.com>
> > > > >> wrote:
> > > > >>
> > > > >> > Hi Stanislav.  See
> > https://tools.ietf.org/html/rfc7628#section-3.1,
> > > > and
> > > > >> > that section refers to the core ABNF productions defined in
> > > > >> > https://tools.ietf.org/html/rfc5234#appendix-B.
> > > > >> >
> > > > >> > Ron
> > > > >> >
> > > > >> > > On Jul 23, 2018, at 1:30 AM, Stanislav Kozlovski <
> > > > >> stanis...@confluent.io>
> > > > >> > wrote:
> > > > >> > >
> > > > >> > > Hey Ron and Rajini,
> > > > >> > >
> > > > >> > > Here are my thoughts:
> > > > >> > > Regarding separators in SaslExtensions - Agreed, that was a
> bad
> > > > move.
> > > > >> > > Should definitely not be a concern of CallbackHandler and
> > > > LoginModule
> > > > >> > > implementors.
> > > > >> > > SaslExtensions interface - Wouldn't implementing it as an
> > > interface
> > > > >> mean
> > > > >> > > that users will have to make sure they're passing in an
> > > unmodifiable
> > > > >> map

Re: [VOTE] KIP-342 - Add support for custom SASL extensions in OAuthBearer authentication

[Ron Dagostino]

+1 (Non-binding).  Thanks for the KIP and the PR, Stanislav.

Ron

On Wed, Jul 25, 2018 at 1:04 PM Stanislav Kozlovski 
wrote:

> Hey everbody,
>
> I'd like to start a vote thread for KIP-342 Add support for custom SASL
> extensions in OAuthBearer authentication
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-342%3A+Add+support+for+Custom+SASL+extensions+in+OAuthBearer+authentication
> >
>
> --
> Best,
> Stanislav
>

Re: [DISCUSS] KIP-346 - Limit blast radius of log compaction failure

[Colin McCabe]

On Mon, Jul 23, 2018, at 23:20, James Cheng wrote:
> Hi Stanislav! Thanks for this KIP!
> 
> I agree that it would be good if the LogCleaner were more tolerant of 
> errors. Currently, as you said, once it dies, it stays dead. 
> 
> Things are better now than they used to be. We have the metric
>   kafka.log:type=LogCleanerManager,name=time-since-last-run-ms
> which we can use to tell us if the threads are dead. And as of 1.1.0, we 
> have KIP-226, which allows you to restart the log cleaner thread, 
> without requiring a broker restart. 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-226+-+Dynamic+Broker+Configuration
>  
> 
>  
> I've only read about this, I haven't personally tried it.

Thanks for pointing this out, James!  Stanislav, we should probably add a 
sentence or two mentioning the KIP-226 changes somewhere in the KIP.  Maybe in 
the intro section?

I think it's clear that requiring the users to manually restart the log cleaner 
is not a very good solution.  But it's good to know that it's a possibility on 
some older releases.

> 
> Some comments:
> * I like the idea of having the log cleaner continue to clean as many 
> partitions as it can, skipping over the problematic ones if possible.
> 
> * If the log cleaner thread dies, I think it should automatically be 
> revived. Your KIP attempts to do that by catching exceptions during 
> execution, but I think we should go all the way and make sure that a new 
> one gets created, if the thread ever dies.

This is inconsistent with the way the rest of Kafka works.  We don't 
automatically re-create other threads in the broker if they terminate.  In 
general, if there is a serious bug in the code, respawning threads is likely to 
make things worse, by putting you in an infinite loop which consumes resources 
and fires off continuous log messages.

> 
> * It might be worth trying to re-clean the uncleanable partitions. I've 
> seen cases where an uncleanable partition later became cleanable. I 
> unfortunately don't remember how that happened, but I remember being 
> surprised when I discovered it. It might have been something like a 
> follower was uncleanable but after a leader election happened, the log 
> truncated and it was then cleanable again. I'm not sure.

James, I disagree.  We had this behavior in the Hadoop Distributed File System 
(HDFS) and it was a constant source of user problems.

What would happen is disks would just go bad over time.  The DataNode would 
notice this and take them offline.  But then, due to some "optimistic" code, 
the DataNode would periodically try to re-add them to the system.  Then one of 
two things would happen: the disk would just fail immediately again, or it 
would appear to work and then fail after a short amount of time.

The way the disk failed was normally having an I/O request take a really long 
time and time out.  So a bunch of request handler threads would basically slam 
into a brick wall when they tried to access the bad disk, slowing the DataNode 
to a crawl.  It was even worse in the second scenario, if the disk appeared to 
work for a while, but then failed.  Any data that had been written on that 
DataNode to that disk would be lost, and we would need to re-replicate it.

Disks aren't biological systems-- they don't heal over time.  Once they're bad, 
they stay bad.  The log cleaner needs to be robust against cases where the disk 
really is failing, and really is returning bad data or timing out.

> 
> * For your metrics, can you spell out the full metric in JMX-style 
> format, such as:
>   kafka.log:type=LogCleanerManager,name=uncleanable-partitions-count
>   value=4
> 
> * For "uncleanable-partitions": topic-partition names can be very long. 
> I think the current max size is 210 characters (or maybe 240-ish?). 
> Having the "uncleanable-partitions" being a list could be very large 
> metric. Also, having the metric come out as a csv might be difficult to 
> work with for monitoring systems. If we *did* want the topic names to be 
> accessible, what do you think of having the 
>   kafka.log:type=LogCleanerManager,topic=topic1,partition=2
> I'm not sure if LogCleanerManager is the right type, but my example was 
> that the topic and partition can be tags in the metric. That will allow 
> monitoring systems to more easily slice and dice the metric. I'm not 
> sure what the attribute for that metric would be. Maybe something like  
> "uncleaned bytes" for that topic-partition? Or time-since-last-clean? Or 
> maybe even just "Value=1".

I haven't though about this that hard, but do we really need the uncleanable 
topic names to be accessible through a metric?  It seems like the admin should 
notice that uncleanable partitions are present, and then check the logs?

> 
> * About `max.uncleanable.partitions`, you said that this likely 
> indicates that the disk is having 

Re: [VOTE] KIP-344: The auto-generated client id should be passed to MetricsReporter

[Ted Yu]

+1

On Wed, Jul 25, 2018 at 9:49 AM Kevin Lu  wrote:

> Hi All,
>
> I am calling a vote for KIP-344
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-344%3A+The+auto-generated+client+id+should+be+passed+to+MetricsReporter
> .
>
> If any committer feels that this minor change does not need to go through
> the KIP process, then feel free to leave a code review at
> https://github.com/apache/kafka/pull/5383 and I will cancel the KIP.
>
> Regards,
> Kevin
>

[jira] [Resolved] (KAFKA-5799) New KafkaSpoutConfig(Scheme)-ByteArrayKeyValueScheme

[Manikumar (JIRA)]

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

Manikumar resolved KAFKA-5799.
--
Resolution: Auto Closed

Closing Apache Storm - Kafka Spout related query.  If this still issue, please 
contact storm mailing list.

> New KafkaSpoutConfig(Scheme)-ByteArrayKeyValueScheme
> 
>
> Key: KAFKA-5799
> URL: https://issues.apache.org/jira/browse/KAFKA-5799
> Project: Kafka
>  Issue Type: New Feature
>Affects Versions: 0.11.0.0
> Environment: apache-storm 1.1.0
>Reporter: Juhong NamGung
>Priority: Minor
> Attachments: 1.JPG, 2.JPG, bakvs.JPG
>
>
> I try to integrate Kafka with Apache Strom.
> I want to get data from Kafka, using KafkaSpout in Apache Storm. 
> To get data from Kafka using KafkaSpout, SpoutConfig-scheme must be setting. 
> (Scheme is an interface that dictates how the ByteBuffer consumed from Kafka 
> gets transformed into a storm tuple)
> I want to get both key and value in Kafka, so I used to KafkaSpoutConfig 
> ‘KeyValueSchemeAsMultiScheme’.
> KeyValueSchemeAsMultiScheme’s Constructor is as follows.
> [^2.JPG]
> But, as you can see in the picture, implementing classes of Interface 
> KeyValueScheme are only StringKeyValueScheme.
> [^1.JPG]
> Using StringKeyValueShceme causes problems when importing Integer data from 
> Kafka. Because StringKeyValueScheme deserialize Bytebuffer to String.
> So I implement ByteArrayKeyValueScheme that deserialize ByteBuffer to 
> ByteArray.
> ByteArrayKeyValueScheme imports data as BtyeArray.
> If you use ByteArrayKeyValueScheme, you can import data regardless of data 
> type from Kafka without error.
> (But, you should convert data type ByteArray to data type that you want(e.g. 
> String, Integer...))
> [^bakvs.JPG]
> {code:java}
> // Some comments here
> import java.nio.ByteBuffer;
> import java.util.List;
> import org.apache.storm.kafka.KeyValueScheme;
> import org.apache.storm.spout.RawScheme;
> import org.apache.storm.tuple.Values;
> import com.google.common.collect.ImmutableMap;
> public class ByteArrayKeyValueScheme extends RawScheme implements 
> KeyValueScheme {
>   @Override
>   public List deserializeKeyAndValue(ByteBuffer key, ByteBuffer 
> value) {
>   // TODO Auto-generated method stub
>   if (key == null) {
>   return deserialize(value);
>   }
>   Object keytuple = deserialize(key).get(0);
>   Object valuetuple = deserialize(value).get(0);
>   return new Values(ImmutableMap.of(keytuple, valuetuple));
>   }
> }
> {code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [Discuss] KIP-321: Add method to get TopicNameExtractor in TopologyDescription

[Guozhang Wang]

Thanks Nishanth,

I've taken a look at the updated KIP and it looks good to me. I think you
can start a new VOTE thread on the current proposal.


Guozhang

On Tue, Jul 24, 2018 at 5:56 PM, Nishanth Pradeep 
wrote:

> I have updated the KIP
>  TopologyDescription+to+better+represent+Source+and+Sink+Nodes>
> .
>
> Changes to the KIP:
>
>- Removed topics() from the Public Interface and Proposed Changes
>sections.
>- Added topics() to the Deprecation plan.
>
> Thanks again for the feedback.
>
> Best,
> Nishanth Pradeep
>
> On Tue, Jul 24, 2018 at 11:21 AM Guozhang Wang  wrote:
>
> > We should not remove it immediately in the up coming 2.1 release. Usually
> > we first mark an API as deprecated, and consider removing it only after
> it
> > has been deprecated for at least one major release period.
> >
> >
> > Guozhang
> >
> > On Mon, Jul 23, 2018 at 7:40 PM, Nishanth Pradeep  >
> > wrote:
> >
> > > Sounds good to me too.
> > >
> > > As far as deprecating goes -- should the topics() method removed
> > completely
> > > or should it have a @deprecated annotation for removal in some future
> > > version?
> > >
> > > Best,
> > > Nishanth Pradeep
> > >
> > > On Sun, Jul 22, 2018 at 1:32 PM Matthias J. Sax  >
> > > wrote:
> > >
> > > > Works for me.
> > > >
> > > > On 7/22/18 9:48 AM, Guozhang Wang wrote:
> > > > > I think I can be convinced with deprecating topics() to keep API
> > > minimal.
> > > > >
> > > > > About renaming the others with `XXNames()`: well, to me it feels
> > still
> > > > not
> > > > > very worthy since although it is not a big burden, it seems also
> not
> > a
> > > > big
> > > > > "return" if we name the newly added function `topicSet()`.
> > > > >
> > > > >
> > > > > Guozhang
> > > > >
> > > > >
> > > > > On Fri, Jul 20, 2018 at 7:38 PM, Nishanth Pradeep <
> > > nishanth...@gmail.com
> > > > >
> > > > > wrote:
> > > > >
> > > > >> I definitely agree with you on deprecating topics().
> > > > >>
> > > > >> I also think changing the method names for consistency is
> > reasonable,
> > > > since
> > > > >> there is no functionality change. Although, I can be convinced
> > either
> > > > way
> > > > >> on this one.
> > > > >>
> > > > >> Best,
> > > > >> Nishanth Pradeep
> > > > >> On Fri, Jul 20, 2018 at 12:15 PM Matthias J. Sax <
> > > matth...@confluent.io
> > > > >
> > > > >> wrote:
> > > > >>
> > > > >>> I would still deprecate existing `topics()` method. If users
> need a
> > > > >>> String, they can call `topicSet().toString()`.
> > > > >>>
> > > > >>> It's just a personal preference, because I believe it's good to
> > keep
> > > > the
> > > > >>> API "minimal".
> > > > >>>
> > > > >>> About renaming the other methods: I thinks it's a very small
> burden
> > > to
> > > > >>> deprecate the existing methods and add them with new names. Also
> > just
> > > > my
> > > > >>> 2 cents.
> > > > >>>
> > > > >>> Would be good to see what others think.
> > > > >>>
> > > > >>>
> > > > >>> -Matthias
> > > > >>>
> > > > >>> On 7/19/18 6:20 PM, Nishanth Pradeep wrote:
> > > >  Understood, Guozhang.
> > > > 
> > > >  Thanks for the help, everyone! I have updated the KIP. Let me
> know
> > > if
> > > > >> you
> > > >  any other thoughts or suggestions.
> > > > 
> > > >  Best,
> > > >  Nishanth Pradeep
> > > > 
> > > >  On Thu, Jul 19, 2018 at 7:33 PM Guozhang Wang <
> wangg...@gmail.com
> > >
> > > > >>> wrote:
> > > > 
> > > > > I see.
> > > > >
> > > > > Well, I think if we add a new function like topicSet() it is
> less
> > > > >>> needed to
> > > > > deprecate topics() as it returns "{topic1, topic2, ..}" which
> is
> > > sort
> > > > >> of
> > > > > non-overlapping in usage with the new API.
> > > > >
> > > > >
> > > > > Guozhang
> > > > >
> > > > > On Thu, Jul 19, 2018 at 5:31 PM, Nishanth Pradeep <
> > > > >>> nishanth...@gmail.com>
> > > > > wrote:
> > > > >
> > > > >> That is what I meant. I will add topicSet() instead of
> changing
> > > the
> > > > >> signature of topics() for compatibility reasons. But should we
> > not
> > > > >> add
> > > > >>> a
> > > > >> @deprecated flag for topics() or do you want to keep it around
> > for
> > > > >> the
> > > > > long
> > > > >> run?
> > > > >>
> > > > >> On Thu, Jul 19, 2018 at 7:27 PM Guozhang Wang <
> > wangg...@gmail.com
> > > >
> > > > > wrote:
> > > > >>
> > > > >>> We cannot change the signature of the function named "topics"
> > > from
> > > > >> "String"
> > > > >>> to "Set", as Matthias mentioned it is a compatibility
> > > > >> breaking
> > > > >>> change.
> > > > >>>
> > > > >>> That's why I was proposing add a new function like
> "Set
> > > > >>> topicSet()", while keeping "String topics()" as is.
> > > > >>>
> > > > >>> Guozhang
> > > > >>>
> > > > >>> On Thu, Jul 19, 2018 at 5:22 PM, Nishanth 

Re: Request wiki access to create KIP

[Guozhang Wang]

Hello Shaobo,

Thanks for your interest in contributing. I've granted you the permission.


Cheers,

Guozhang

On Wed, Jul 25, 2018 at 7:20 AM, lambdaliu(刘少波) 
wrote:

> Hi,
>
> Can someone give me the permissions for creating a new KIP?
> My username is *lambdaliu*.
>
> Thanks.
>
> --
> Lambdaliu(Shaobo Liu)




-- 
-- Guozhang

[VOTE] KIP-342 - Add support for custom SASL extensions in OAuthBearer authentication

[Stanislav Kozlovski]

Hey everbody,

I'd like to start a vote thread for KIP-342 Add support for custom SASL
extensions in OAuthBearer authentication


-- 
Best,
Stanislav

Re: [VOTE] 2.0.0 RC3

[Ron Dagostino]

+1 (non-binding)

Built from source and exercised the new SASL/OAUTHBEARER functionality with
unsecured tokens.

Thanks, Rajini -- apologies for KAFKA-7182.

Ron

On Tue, Jul 24, 2018 at 5:10 PM Vahid S Hashemian 
wrote:

> +1 (non-binding)
>
> Built from source and ran quickstart successfully with both Java 8 and
> Java 9 on Ubuntu.
> Thanks Rajini!
>
> --Vahid
>
>
>
>
> From:   Rajini Sivaram 
> To: dev , Users ,
> kafka-clients 
> Date:   07/24/2018 08:33 AM
> Subject:[VOTE] 2.0.0 RC3
>
>
>
> Hello Kafka users, developers and client-developers,
>
>
> This is the fourth candidate for release of Apache Kafka 2.0.0.
>
>
> This is a major version release of Apache Kafka. It includes 40 new  KIPs
> and
>
> several critical bug fixes. Please see the 2.0.0 release plan for more
> details:
>
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=80448820
>
>
>
> A few notable highlights:
>
>- Prefixed wildcard ACLs (KIP-290), Fine grained ACLs for CreateTopics
>(KIP-277)
>- SASL/OAUTHBEARER implementation (KIP-255)
>- Improved quota communication and customization of quotas (KIP-219,
>KIP-257)
>- Efficient memory usage for down conversion (KIP-283)
>- Fix log divergence between leader and follower during fast leader
>failover (KIP-279)
>- Drop support for Java 7 and remove deprecated code including old
> scala
>clients
>- Connect REST extension plugin, support for externalizing secrets and
>improved error handling (KIP-285, KIP-297, KIP-298 etc.)
>- Scala API for Kafka Streams and other Streams API improvements
>(KIP-270, KIP-150, KIP-245, KIP-251 etc.)
>
>
> Release notes for the 2.0.0 release:
>
> http://home.apache.org/~rsivaram/kafka-2.0.0-rc3/RELEASE_NOTES.html
>
>
>
> *** Please download, test and vote by Friday July 27, 4pm PT.
>
>
> Kafka's KEYS file containing PGP keys we use to sign the release:
>
> http://kafka.apache.org/KEYS
>
>
>
> * Release artifacts to be voted upon (source and binary):
>
> http://home.apache.org/~rsivaram/kafka-2.0.0-rc3/
>
>
>
> * Maven artifacts to be voted upon:
>
> https://repository.apache.org/content/groups/staging/
>
>
>
> * Javadoc:
>
> http://home.apache.org/~rsivaram/kafka-2.0.0-rc3/javadoc/
>
>
>
> * Tag to be voted upon (off 2.0 branch) is the 2.0.0 tag:
>
> https://github.com/apache/kafka/releases/tag/2.0.0-rc3
>
>
> * Documentation:
>
> http://kafka.apache.org/20/documentation.html
>
>
>
> * Protocol:
>
> http://kafka.apache.org/20/protocol.html
>
>
>
> * Successful Jenkins builds for the 2.0 branch:
>
> Unit/integration tests:
> https://builds.apache.org/job/kafka-2.0-jdk8/90/
>
>
> System tests:
> https://jenkins.confluent.io/job/system-test-kafka/job/2.0/41/
>
>
>
> /**
>
>
> Thanks,
>
>
>
> Rajini
>
>
>
>
>

[VOTE] KIP-344: The auto-generated client id should be passed to MetricsReporter

[Kevin Lu]

Hi All,

I am calling a vote for KIP-344
https://cwiki.apache.org/confluence/display/KAFKA/KIP-344%3A+The+auto-generated+client+id+should+be+passed+to+MetricsReporter
.

If any committer feels that this minor change does not need to go through
the KIP process, then feel free to leave a code review at
https://github.com/apache/kafka/pull/5383 and I will cancel the KIP.

Regards,
Kevin

Re: Didn't receive video stream data into spark through kafka

[Biswajit Ghosh]

Hi,

Thank you for your support! everything is working except can't figure out
how to pass the video frame( serialize in a pickle file) to spark.

My problem is during loading the pickle file stream got an EOF error into
spark stream context. I'm suspecting that due to large file size TCP broke
that file into some chunk and while receiving in spark, got an RDD of a
single chunk which is not a valid pickle file.

I'm wondering how to get a complete pickle file from kafka to spark.

Thank you!


On 25 July 2018 at 14:51, Attila Sasvári  wrote:

> Hi Biswajit,
>
> Can you please provide more information:
>
> - What other symptoms do you see? Are all your Kafka brokers up and
> running?
> - What replication did you set for offsets.topic.replication.factor (i.e.
> replication factor of __consumer_offsets) in your Kafka broker's config ?
> What is set for min.insync.replicas on the __consumer_offsets topic?
> - What does the following command show? kafka-topics.sh --describe
> --zookeeper 
> - What is the generated consumer group id  (that is used to select the
> consumer group coordinator broker) of your pyspark client? Is it different
> from the one used by kafka console consumer? How do you commit the consumer
> offsets? I suspect that the consumer offset for the consumer group might
> have already been established in Kafka and that is why you are not able to
> get records from your pyspark application. Can you try to run your
> application using a consumer group id that did not exists before in your
> cluster?
> - What is the version of Kafka you are using (broker and
> spark-streaming-kafka)?
>
> Regards,
> Attila
>
> On Tue, Jul 24, 2018 at 3:51 PM Biswajit Ghosh <
> biswaji...@aqbsolutions.com>
> wrote:
>
> > Yes, I have double check that.
> >
> > On 24 July 2018 at 19:20, Aman Rastogi  wrote:
> >
> > > Is your topic same in both the case?
> > >
> > > On Tue, 24 Jul 2018, 19:15 Biswajit Ghosh, <
> biswaji...@aqbsolutions.com>
> > > wrote:
> > >
> > > > Hi team,
> > > >
> > > > I got an issue while integrating with the spark streaming using
> > pyspark,
> > > I
> > > > did receive the video stream data in a different consumer subscribe
> to
> > > the
> > > > same topic.
> > > >
> > > > Works fine with this command : *./kafka-console-consumer.sh
> > > > --bootstrap-server :9092 --topic spark-streaming-consumer
> > > > --from-beginning*
> > > >
> > > > But not with this :
> > > >
> > > > ​
> > > > >
> > > > def processRecord(record):
> > > >
> > > > print(record)
> > > >
> > > >
> > > > > ​sc = SparkContext(master="local[2]",appName="HNStreaming")
> > > >
> > > > ​sc.setLogLevel('DEBUG')
> > > >
> > > > ssc = StreamingContext(sc, 2)
> > > >
> > > > topic = "spark-stream-message"
> > > >
> > > > kvs = KafkaUtils.createDirectStream(ssc, [topic],
> > > {'metadata.broker.list':
> > > > > brokers})
> > > >
> > > > kvs.foreachRDD(self.processRecord)
> > > >
> > > > ssc.start()
> > > >
> > > > ssc.awaitTermination()
> > > >
> > > > ​
> > > > >
> > > > >
> > > >
> > > > Expecting help from your side asap.
> > > >
> > > > Thank you!
> > > >
> > > >
> > > > --
> > > >
> > > > Regards,
> > > > biswajitGhosh
> > > >
> > >
> >
> >
> >
> > --
> >
> > Regards,
> > biswajitGhosh
> >
>



-- 

Regards,
biswajitGhosh

Request wiki access to create KIP

[刘少波]

Hi,

Can someone give me the permissions for creating a new KIP?
My username is *lambdaliu*.

Thanks.

--
Lambdaliu(Shaobo Liu)

[jira] [Resolved] (KAFKA-2396) I have config host.name and advertised.host.name but still have a link connet to 0.0.0.0 with random port

[Manikumar (JIRA)]

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

Manikumar resolved KAFKA-2396.
--
Resolution: Not A Problem

extra port is related to JMX. See  KAFKA-1869

> I have config host.name and advertised.host.name but still have a link connet 
> to 0.0.0.0 with random port
> -
>
> Key: KAFKA-2396
> URL: https://issues.apache.org/jira/browse/KAFKA-2396
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.8.2.0
> Environment: kafka 0.8.2.0 scalas 2.10
>Reporter: lmc
>Priority: Major
>
> while running kafka service, many port have been open, I have config 
> host.name and advertised.host.name in server.propertise,but I found there is 
> a link link to 0.0.0.0 with a random port which I mark with a red color 
> follow,Can you help me that why this random port will be  opened, and please 
> thell me how to set this ip address to what I want.
>  
> >jps
> 7182  kafka
> >netstat -anp|grep 7182
> tcp0  0 0.0.0.0:58115   0.0.0.0:*   LISTEN
>   13927/java  
> tcp0  0 127.0.0.1:9092  0.0.0.0:*   LISTEN
>   13927/java  
> tcp0  0 127.0.0.1:51626 127.0.0.1:2181  
> ESTABLISHED 13927/java 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

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

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-7144: Fix task assignment to be even (#5390)

--
[...truncated 108.95 KB...]
kafka.zk.KafkaZkClientTest > testGetBrokerMethods PASSED

kafka.zk.KafkaZkClientTest > testCreateTokenChangeNotification STARTED

kafka.zk.KafkaZkClientTest > testCreateTokenChangeNotification PASSED

kafka.zk.KafkaZkClientTest > testGetTopicsAndPartitions STARTED

kafka.zk.KafkaZkClientTest > testGetTopicsAndPartitions PASSED

kafka.zk.KafkaZkClientTest > testRegisterBrokerInfo STARTED

kafka.zk.KafkaZkClientTest > testRegisterBrokerInfo PASSED

kafka.zk.KafkaZkClientTest > testConsumerOffsetPath STARTED

kafka.zk.KafkaZkClientTest > testConsumerOffsetPath PASSED

kafka.zk.KafkaZkClientTest > testControllerManagementMethods STARTED

kafka.zk.KafkaZkClientTest > testControllerManagementMethods PASSED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods STARTED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods PASSED

kafka.zk.KafkaZkClientTest > testPropagateIsrChanges STARTED

kafka.zk.KafkaZkClientTest > testPropagateIsrChanges PASSED

kafka.zk.KafkaZkClientTest > testControllerEpochMethods STARTED

kafka.zk.KafkaZkClientTest > testControllerEpochMethods PASSED

kafka.zk.KafkaZkClientTest > testDeleteRecursive STARTED

kafka.zk.KafkaZkClientTest > testDeleteRecursive PASSED

kafka.zk.KafkaZkClientTest > testGetTopicPartitionStates STARTED

kafka.zk.KafkaZkClientTest > testGetTopicPartitionStates PASSED

kafka.zk.KafkaZkClientTest > testCreateConfigChangeNotification STARTED

kafka.zk.KafkaZkClientTest > testCreateConfigChangeNotification PASSED

kafka.zk.KafkaZkClientTest > testDelegationTokenMethods STARTED

kafka.zk.KafkaZkClientTest > testDelegationTokenMethods PASSED

kafka.zk.LiteralAclStoreTest > shouldHaveCorrectPaths STARTED

kafka.zk.LiteralAclStoreTest > shouldHaveCorrectPaths PASSED

kafka.zk.LiteralAclStoreTest > shouldRoundTripChangeNode STARTED

kafka.zk.LiteralAclStoreTest > shouldRoundTripChangeNode PASSED

kafka.zk.LiteralAclStoreTest > shouldThrowFromEncodeOnNoneLiteral STARTED

kafka.zk.LiteralAclStoreTest > shouldThrowFromEncodeOnNoneLiteral PASSED

kafka.zk.LiteralAclStoreTest > shouldWriteChangesToTheWritePath STARTED

kafka.zk.LiteralAclStoreTest > shouldWriteChangesToTheWritePath PASSED

kafka.zk.LiteralAclStoreTest > shouldHaveCorrectPatternType STARTED

kafka.zk.LiteralAclStoreTest > shouldHaveCorrectPatternType PASSED

kafka.zk.LiteralAclStoreTest > shouldDecodeResourceUsingTwoPartLogic STARTED

kafka.zk.LiteralAclStoreTest > shouldDecodeResourceUsingTwoPartLogic PASSED

kafka.zk.ExtendedAclStoreTest > shouldHaveCorrectPaths STARTED

kafka.zk.ExtendedAclStoreTest > shouldHaveCorrectPaths PASSED

kafka.zk.ExtendedAclStoreTest > shouldRoundTripChangeNode STARTED

kafka.zk.ExtendedAclStoreTest > shouldRoundTripChangeNode PASSED

kafka.zk.ExtendedAclStoreTest > shouldThrowFromEncodeOnLiteral STARTED

kafka.zk.ExtendedAclStoreTest > shouldThrowFromEncodeOnLiteral PASSED

kafka.zk.ExtendedAclStoreTest > shouldThrowIfConstructedWithLiteral STARTED

kafka.zk.ExtendedAclStoreTest > shouldThrowIfConstructedWithLiteral PASSED

kafka.zk.ExtendedAclStoreTest > shouldWriteChangesToTheWritePath STARTED

kafka.zk.ExtendedAclStoreTest > shouldWriteChangesToTheWritePath PASSED

kafka.zk.ExtendedAclStoreTest > shouldHaveCorrectPatternType STARTED

kafka.zk.ExtendedAclStoreTest > shouldHaveCorrectPatternType PASSED

kafka.server.LogOffsetTest > testFetchOffsetsBeforeWithChangingSegmentSize 
STARTED

kafka.server.LogOffsetTest > testFetchOffsetsBeforeWithChangingSegmentSize 
PASSED

kafka.server.LogOffsetTest > testGetOffsetsBeforeEarliestTime STARTED

kafka.server.LogOffsetTest > testGetOffsetsBeforeEarliestTime PASSED

kafka.server.LogOffsetTest > testGetOffsetsForUnknownTopic STARTED

kafka.server.LogOffsetTest > testGetOffsetsForUnknownTopic PASSED

kafka.server.LogOffsetTest > testEmptyLogsGetOffsets STARTED

kafka.server.LogOffsetTest > testEmptyLogsGetOffsets PASSED

kafka.server.LogOffsetTest > testFetchOffsetsBeforeWithChangingSegments STARTED
ERROR: Could not install GRADLE_4_8_1_HOME
java.lang.NullPointerException
at 
hudson.plugins.toolenv.ToolEnvBuildWrapper$1.buildEnvVars(ToolEnvBuildWrapper.java:46)
at hudson.model.AbstractBuild.getEnvironment(AbstractBuild.java:895)
at hudson.plugins.git.GitSCM.getParamExpandedRepos(GitSCM.java:483)
at 
hudson.plugins.git.GitSCM.compareRemoteRevisionWithImpl(GitSCM.java:692)
at hudson.plugins.git.GitSCM.compareRemoteRevisionWith(GitSCM.java:657)
at hudson.scm.SCM.compareRemoteRevisionWith(SCM.java:400)
at hudson.scm.SCM.poll(SCM.java:417)
at hudson.model.AbstractProject._poll(AbstractProject.java:1390)
at hudson.model.AbstractProject.poll(AbstractProject.java:1293)
at 

[jira] [Resolved] (KAFKA-2395) Linked to a interfaces even though the host.name had config

[Manikumar (JIRA)]

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

Manikumar resolved KAFKA-2395.
--
Resolution: Not A Problem

extra port is related to KAFKA-1869

> Linked to  a interfaces even though the host.name had config
> 
>
> Key: KAFKA-2395
> URL: https://issues.apache.org/jira/browse/KAFKA-2395
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.8.2.0
> Environment: kafka_2.10-0.8.2.0
>Reporter: lmc
>Priority: Major
>
> while running kafka service, many port have been open, I have config 
> host.name and advertised.host.name in server.propertise,but I found there is 
> a link link to 0.0.0.0 with a random port which I mark with a red color 
> follow,Can you help me that why this random port will be  opened, and please 
> thell me how to set this ip address to what I want.
>  
> >jps
> 7182  kafka
> >netstat -anp|grep 7182
> tcp0  0 0.0.0.0:58115   0.0.0.0:*   LISTEN
>   13927/java  
> tcp0  0 127.0.0.1:9092  0.0.0.0:*   LISTEN
>   13927/java  
> tcp0  0 127.0.0.1:51626 127.0.0.1:2181  
> ESTABLISHED 13927/java 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: Didn't receive video stream data into spark through kafka

[Attila Sasvári]

Hi Biswajit,

Can you please provide more information:

- What other symptoms do you see? Are all your Kafka brokers up and
running?
- What replication did you set for offsets.topic.replication.factor (i.e.
replication factor of __consumer_offsets) in your Kafka broker's config ?
What is set for min.insync.replicas on the __consumer_offsets topic?
- What does the following command show? kafka-topics.sh --describe
--zookeeper 
- What is the generated consumer group id  (that is used to select the
consumer group coordinator broker) of your pyspark client? Is it different
from the one used by kafka console consumer? How do you commit the consumer
offsets? I suspect that the consumer offset for the consumer group might
have already been established in Kafka and that is why you are not able to
get records from your pyspark application. Can you try to run your
application using a consumer group id that did not exists before in your
cluster?
- What is the version of Kafka you are using (broker and
spark-streaming-kafka)?

Regards,
Attila

On Tue, Jul 24, 2018 at 3:51 PM Biswajit Ghosh 
wrote:

> Yes, I have double check that.
>
> On 24 July 2018 at 19:20, Aman Rastogi  wrote:
>
> > Is your topic same in both the case?
> >
> > On Tue, 24 Jul 2018, 19:15 Biswajit Ghosh, 
> > wrote:
> >
> > > Hi team,
> > >
> > > I got an issue while integrating with the spark streaming using
> pyspark,
> > I
> > > did receive the video stream data in a different consumer subscribe to
> > the
> > > same topic.
> > >
> > > Works fine with this command : *./kafka-console-consumer.sh
> > > --bootstrap-server :9092 --topic spark-streaming-consumer
> > > --from-beginning*
> > >
> > > But not with this :
> > >
> > > ​
> > > >
> > > def processRecord(record):
> > >
> > > print(record)
> > >
> > >
> > > > ​sc = SparkContext(master="local[2]",appName="HNStreaming")
> > >
> > > ​sc.setLogLevel('DEBUG')
> > >
> > > ssc = StreamingContext(sc, 2)
> > >
> > > topic = "spark-stream-message"
> > >
> > > kvs = KafkaUtils.createDirectStream(ssc, [topic],
> > {'metadata.broker.list':
> > > > brokers})
> > >
> > > kvs.foreachRDD(self.processRecord)
> > >
> > > ssc.start()
> > >
> > > ssc.awaitTermination()
> > >
> > > ​
> > > >
> > > >
> > >
> > > Expecting help from your side asap.
> > >
> > > Thank you!
> > >
> > >
> > > --
> > >
> > > Regards,
> > > biswajitGhosh
> > >
> >
>
>
>
> --
>
> Regards,
> biswajitGhosh
>

[jira] [Resolved] (KAFKA-7159) mark configuration files in confluent-kafka RPM SPEC file

[Attila Sasvari (JIRA)]

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

Attila Sasvari resolved KAFKA-7159.
---
Resolution: Won't Fix

> mark configuration files in confluent-kafka RPM SPEC file
> -
>
> Key: KAFKA-7159
> URL: https://issues.apache.org/jira/browse/KAFKA-7159
> Project: Kafka
>  Issue Type: Improvement
>  Components: packaging
>Affects Versions: 1.1.0
> Environment: RHEL7
>Reporter: Robert Fabisiak
>Priority: Trivial
>  Labels: rpm
>
> All configuration files in kafka RPM SPEC file should be marked with %config 
> prefix in %files section.
> This would prevent overwrites during install/upgrade and uninstall operations
> [https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/7/html-single/rpm_packaging_guide/index#files]
> It's especially important to save configuration during package upgrades.
> Section to change in SPEC file:
> {code:java}
> %files
> %config(noreplace) %{_sysconfdir}/kafka/*.conf
> %config(noreplace) %{_sysconfdir}/kafka/*.properties
> {code}
> It would also be good to mark documentation files with %doc



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Build failed in Jenkins: kafka-trunk-jdk10 #315

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-7144: Fix task assignment to be even (#5390)

--
[...truncated 1.98 MB...]
org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldNotAllowToAddGlobalStoreWithSourceNameEqualsProcessorName STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldNotAllowToAddGlobalStoreWithSourceNameEqualsProcessorName PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddSourceWithOffsetReset STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddSourceWithOffsetReset PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddStateStoreWithSource STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddStateStoreWithSource PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddInternalTopicConfigForRepartitionTopics STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddInternalTopicConfigForRepartitionTopics PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldNotAllowNullProcessorNameWhenConnectingProcessorAndStateStores STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldNotAllowNullProcessorNameWhenConnectingProcessorAndStateStores PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddSourceWithSameName STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddSourceWithSameName PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddMoreThanOnePatternSourceNode STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddMoreThanOnePatternSourceNode PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddTimestampExtractorWithPatternPerSource STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddTimestampExtractorWithPatternPerSource PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddProcessorWithSameName STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddProcessorWithSameName PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldSetCorrectSourceNodesWithRegexUpdatedTopics STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldSetCorrectSourceNodesWithRegexUpdatedTopics PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldConnectRegexMatchedTopicsToStateStore STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldConnectRegexMatchedTopicsToStateStore PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddSourceWithSameTopic STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddSourceWithSameTopic PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testTopicGroupsByStateStore STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testTopicGroupsByStateStore PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddSourceWithoutOffsetReset STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddSourceWithoutOffsetReset PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddStateStoreWithDuplicates STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testAddStateStoreWithDuplicates PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldNotAllowNullStateStoreNameWhenConnectingProcessorAndStateStores STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldNotAllowNullStateStoreNameWhenConnectingProcessorAndStateStores PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddTimestampExtractorPerSource STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAddTimestampExtractorPerSource PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testPatternSourceTopic STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
testPatternSourceTopic PASSED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 
shouldAssociateStateStoreNameWhenStateStoreSupplierIsExternal STARTED

org.apache.kafka.streams.processor.internals.InternalTopologyBuilderTest > 

Metric names

[Mathias Kub]

Hi there,

I am having problems aggregating metrics for our monitoring system,
because some JMX beans use dots in its names (e.g.
kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec, "topic" just
uses the plain topic name as is).

Other JMX beans replace dots by underscores, like the one added in the PR.

What is the problem when using dots in the topic name there? At least in
our case, everything seems to work fine, even though our topics do
contain dots.

However, the current behavior is different between these metrics which
makes combining the metrics hard.

Please also see
https://github.com/apache/kafka/pull/4362#discussion_r203949516.

Thanks,
Mathias

***UNCHECKED*** Re: [DISCUSS] KIP-289: Improve the default group id behavior in KafkaConsumer

[Jason Gustafson]

Hey Vahid,

Sorry for the confusion. I think we all agree that going forward, we
shouldn't support the empty group id, so the question is just around
compatibility. I think we have to bump the OffsetCommit API version so that
old clients which are unknowingly depending on the default empty group id
will continue to work with new brokers. For new versions of the client, we
can either drop support for the empty group id immediately or we can give
users a grace period. I was thinking we would do the latter. We can change
the default group.id, but in the case that a user has explicitly configured
the empty group, then we can just use an old version of the OffsetCommit
API which still supports it. In a future release, we can drop this support
and only use the latest OffsetCommit version. Does that make sense?

Thanks,
Jason


On Tue, Jul 24, 2018 at 12:36 PM, Vahid S Hashemian <
vahidhashem...@us.ibm.com> wrote:

> Hi Jason,
>
> Thanks for clarifying.
>
> So if we are going to continue supporting the empty group id as before
> (with only an addition of a deprecation warning), and disable
> enable.auto.commit for the new default (null) group id on the client side,
> do we really need to bump up the OffsetCommit version?
>
> You mentioned "If an explicit empty string is configured for the group id,
> then maybe we keep the current behavior for compatibility" which makes
> sense to me, but I find it in conflict with your earlier suggestion "we
> just need to bump the OffsetCommit request API and only accept the offset
> commit for older versions.". Maybe I'm missing something?
>
> Thanks!
> --Vahid
>
>
>
>
> From:   Jason Gustafson 
> To: dev 
> Date:   07/23/2018 10:52 PM
> Subject:Re: [DISCUSS] KIP-289: Improve the default group id
> behavior in KafkaConsumer
>
>
>
> Hey Vahid,
>
> Thanks for the updates. Just to clarify, I was suggesting that we disable
> enable.auto.commit only if no explicit group.id is configured. If an
> explicit empty string is configured for the group id, then maybe we keep
> the current behavior for compatibility. We can log a warning mentioning
> the
> deprecation and we can use the old version of the OffsetCommit API that
> allows the empty group id. In a later release, we can drop this support in
> the client. Does that seem reasonable?
>
> By the way, instead of using the new ILLEGAL_OFFSET_COMMIT error code,
> couldn't we use INVALID_GROUP_ID?
>
> Thanks,
> Jason
>
>
>
> On Mon, Jul 23, 2018 at 5:14 PM, Stanislav Kozlovski
>  > wrote:
>
> > Hey Vahid,
> >
> > No I don't see an issue with it. I believe it to be the best approach.
> >
> > Best,
> > Stanisav
> >
> > On Mon, Jul 23, 2018 at 12:41 PM Vahid S Hashemian <
> > vahidhashem...@us.ibm.com> wrote:
> >
> > > Hi Stanislav,
> > >
> > > Thanks for the feedback.
> > > Do you see an issue with using `null` as the default group id (as
> > > addressed by Jason in his response)?
> > > This default group id would not support offset commits and consumers
> > would
> > > use `auto.offset.reset` config when there is no current offset.
> > >
> > > Thanks.
> > > --Vahid
> > >
> > >
> > >
> > >
> > > From:   Stanislav Kozlovski 
> > > To: dev@kafka.apache.org
> > > Date:   07/20/2018 11:09 AM
> > > Subject:Re: [DISCUSS] KIP-289: Improve the default group id
> > > behavior in KafkaConsumer
> > >
> > >
> > >
> > > I agree with Jason's notion that
> > > >  implicit use of the empty group.id to commit offsets is more likely
> > to
> > > be causing users unexpected problems than actually providing a useful
> > > capability.
> > > I was initially confused that this is the behavior when investigating
> a
> > > new-ish JIRA issue <
> > >
> https://issues.apache.org/jira/browse/KAFKA-6758
>
> > > > about
> > > the same topic.
> > > So, +1 to deprecating "" as a group.id
> > >
> > > The question after that becomes what the *default* value should be -
> > > should
> > > we:
> > > a) treat an unconfigured group.id consumer as a sort of intermittent
> > > consumer where you don't store offsets at all (thereby making the user
> > > explicitly sign up for them)
> > > b) have a default value which makes use of them? I sort of like the
> > > former.
> > >
> > > @Dhruvil, thinking about it at a high-level - yes. I can't think of a
> > > situation where it makes sense to name something an empty string as
> far
> > as
> > > I'm aware - to me it seems like potential for confusion
> > >
> > >
> > > On Fri, Jul 20, 2018 at 10:22 AM Rajini Sivaram
>  > >
> > > wrote:
> > >
> > > > +1 to deprecate use of "" as group.id since it is odd to have a
> > resource
> > > > name that you cannot set ACLs for. Agree, we have to support older
> > > clients
> > > > though.
> > > >
> > > > Thanks,
> > > >
> > > > Rajini
> > > >
> > > > On Fri, Jul 20, 2018 at 5:25 PM, Jason Gustafson
> 
> > > > wrote:
> > > >
> > > > > Hi Vahid,
> > > > >
> > > > > Sorry for getting to this so late. I think there are two things
> here:
> > > > >
> > > > > 1. The use of "" 

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

[Apache Jenkins Server]

See 


Changes:

[github] MINOR: internal config objects should not be logged (#5389)

--
[...truncated 883.98 KB...]

kafka.utils.SchedulerTest > testMockSchedulerPeriodicTask STARTED

kafka.utils.SchedulerTest > testMockSchedulerPeriodicTask PASSED

kafka.utils.SchedulerTest > testNonPeriodicTask STARTED

kafka.utils.SchedulerTest > testNonPeriodicTask PASSED

kafka.utils.SchedulerTest > testRestart STARTED

kafka.utils.SchedulerTest > testRestart PASSED

kafka.utils.SchedulerTest > testReentrantTaskInMockScheduler STARTED

kafka.utils.SchedulerTest > testReentrantTaskInMockScheduler PASSED

kafka.utils.SchedulerTest > testPeriodicTask STARTED

kafka.utils.SchedulerTest > testPeriodicTask PASSED

kafka.utils.LoggingTest > testLog4jControllerIsRegistered STARTED

kafka.utils.LoggingTest > testLog4jControllerIsRegistered PASSED

kafka.utils.LoggingTest > testLogName STARTED

kafka.utils.LoggingTest > testLogName PASSED

kafka.utils.LoggingTest > testLogNameOverride STARTED

kafka.utils.LoggingTest > testLogNameOverride PASSED

kafka.utils.ZkUtilsTest > testGetSequenceIdMethod STARTED

kafka.utils.ZkUtilsTest > testGetSequenceIdMethod PASSED

kafka.utils.ZkUtilsTest > testAbortedConditionalDeletePath STARTED

kafka.utils.ZkUtilsTest > testAbortedConditionalDeletePath PASSED

kafka.utils.ZkUtilsTest > testGetAllPartitionsTopicWithoutPartitions STARTED

kafka.utils.ZkUtilsTest > testGetAllPartitionsTopicWithoutPartitions PASSED

kafka.utils.ZkUtilsTest > testSuccessfulConditionalDeletePath STARTED

kafka.utils.ZkUtilsTest > testSuccessfulConditionalDeletePath PASSED

kafka.utils.ZkUtilsTest > testPersistentSequentialPath STARTED

kafka.utils.ZkUtilsTest > testPersistentSequentialPath PASSED

kafka.utils.ZkUtilsTest > testClusterIdentifierJsonParsing STARTED

kafka.utils.ZkUtilsTest > testClusterIdentifierJsonParsing PASSED

kafka.utils.ZkUtilsTest > testGetLeaderIsrAndEpochForPartition STARTED

kafka.utils.ZkUtilsTest > testGetLeaderIsrAndEpochForPartition PASSED

kafka.utils.CoreUtilsTest > testGenerateUuidAsBase64 STARTED

kafka.utils.CoreUtilsTest > testGenerateUuidAsBase64 PASSED

kafka.utils.CoreUtilsTest > testAbs STARTED

kafka.utils.CoreUtilsTest > testAbs PASSED

kafka.utils.CoreUtilsTest > testReplaceSuffix STARTED

kafka.utils.CoreUtilsTest > testReplaceSuffix PASSED

kafka.utils.CoreUtilsTest > testCircularIterator STARTED

kafka.utils.CoreUtilsTest > testCircularIterator PASSED

kafka.utils.CoreUtilsTest > testReadBytes STARTED

kafka.utils.CoreUtilsTest > testReadBytes PASSED

kafka.utils.CoreUtilsTest > testCsvList STARTED

kafka.utils.CoreUtilsTest > testCsvList PASSED

kafka.utils.CoreUtilsTest > testReadInt STARTED

kafka.utils.CoreUtilsTest > testReadInt PASSED

kafka.utils.CoreUtilsTest > testAtomicGetOrUpdate STARTED

kafka.utils.CoreUtilsTest > testAtomicGetOrUpdate PASSED

kafka.utils.CoreUtilsTest > testUrlSafeBase64EncodeUUID STARTED

kafka.utils.CoreUtilsTest > testUrlSafeBase64EncodeUUID PASSED

kafka.utils.CoreUtilsTest > testCsvMap STARTED

kafka.utils.CoreUtilsTest > testCsvMap PASSED

kafka.utils.CoreUtilsTest > testInLock STARTED

kafka.utils.CoreUtilsTest > testInLock PASSED

kafka.utils.CoreUtilsTest > testTryAll STARTED

kafka.utils.CoreUtilsTest > testTryAll PASSED

kafka.utils.CoreUtilsTest > testSwallow STARTED

kafka.utils.CoreUtilsTest > testSwallow PASSED

kafka.utils.IteratorTemplateTest > testIterator STARTED

kafka.utils.IteratorTemplateTest > testIterator PASSED

kafka.utils.json.JsonValueTest > testJsonObjectIterator STARTED

kafka.utils.json.JsonValueTest > testJsonObjectIterator PASSED

kafka.utils.json.JsonValueTest > testDecodeLong STARTED

kafka.utils.json.JsonValueTest > testDecodeLong PASSED

kafka.utils.json.JsonValueTest > testAsJsonObject STARTED

kafka.utils.json.JsonValueTest > testAsJsonObject PASSED

kafka.utils.json.JsonValueTest > testDecodeDouble STARTED

kafka.utils.json.JsonValueTest > testDecodeDouble PASSED

kafka.utils.json.JsonValueTest > testDecodeOption STARTED

kafka.utils.json.JsonValueTest > testDecodeOption PASSED

kafka.utils.json.JsonValueTest > testDecodeString STARTED

kafka.utils.json.JsonValueTest > testDecodeString PASSED

kafka.utils.json.JsonValueTest > testJsonValueToString STARTED

kafka.utils.json.JsonValueTest > testJsonValueToString PASSED

kafka.utils.json.JsonValueTest > testAsJsonObjectOption STARTED

kafka.utils.json.JsonValueTest > testAsJsonObjectOption PASSED

kafka.utils.json.JsonValueTest > testAsJsonArrayOption STARTED

kafka.utils.json.JsonValueTest > testAsJsonArrayOption PASSED

kafka.utils.json.JsonValueTest > testAsJsonArray STARTED

kafka.utils.json.JsonValueTest > testAsJsonArray PASSED

kafka.utils.json.JsonValueTest > testJsonValueHashCode STARTED

kafka.utils.json.JsonValueTest > testJsonValueHashCode PASSED

kafka.utils.json.JsonValueTest > testDecodeInt STARTED