[jira] [Commented] (KAFKA-5337) Partition assignment strategy that distributes lag evenly across consumers in each group

[Grant Neale (JIRA)]

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

Grant Neale commented on KAFKA-5337:


PR raised: https://github.com/apache/kafka/pull/3158

> Partition assignment strategy that distributes lag evenly across consumers in 
> each group
> 
>
> Key: KAFKA-5337
> URL: https://issues.apache.org/jira/browse/KAFKA-5337
> Project: Kafka
>  Issue Type: New Feature
>  Components: consumer
>Affects Versions: 0.10.2.1
>Reporter: Grant Neale
>Priority: Minor
>
> Existing partition assignment strategies (RangeAssignor and 
> RoundRobinAssignor) do not account for the current consumer group lag on each 
> partition.  This can result in sub-optimal assignments when the distribution 
> of lags for a given topic and consumer group is skewed.
> The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
> partitions such that lag is distributed as evenly across a consumer group.
> h4. Algorithm:
> For each topic, first obtain the lag on all partitions. Lag on a given 
> partition is the difference between the end offset and the last offset 
> committed by the consumer group. If no offsets have been committed for a 
> partition we determine the lag based on the code auto.offset.reset property. 
> If auto.offset.reset=latest, we assume a lag of 0. If 
> auto.offset.reset=earliest (or any other value) we assume lag equal to the 
> total number of message currently available in that partition.
> Next, create a map storing the current total lag of all partitions assigned 
> to each member of the consumer group. Partitions are assigned in decreasing 
> order of lag, with each partition assigned to the consumer with least total 
> number of assigned partitions, breaking ties by assigning to the consumer 
> with the least total currently assigned lag.
> Assigning partitions evenly across consumers (by partition count) ensures 
> that the assignment is reasonably balanced (by partition count) when all 
> partitions have a current lag of 0 or if the distribution of lags is heavily 
> skewed. It also gives the consumer group the best possible chance of 
> remaining balanced if the assignment is retained for a long period (assuming 
> throughput is consistent across members of the consumer group).



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

[jira] [Commented] (KAFKA-5337) Partition assignment strategy that distributes lag evenly across consumers in each group

[ASF GitHub Bot (JIRA)]

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

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

GitHub user grantneale opened a pull request:

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

KAFKA-5337: LagBasedAssignor partition assignment strategy

Existing partition assignment strategies (RangeAssignor and 
RoundRobinAssignor) do not account for the current consumer group lag on each 
partition. This can result in sub-optimal assignments when the distribution of 
lags for a given topic and consumer group is skewed.

The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
partitions such that lag is distributed as evenly across a consumer group.

## Algorithm

For each topic, first obtain the lag on all partitions. Lag on a given 
partition is the difference between the end offset and the last offset 
committed by the consumer group. If no offsets have been committed for a 
partition we determine the lag based on the code auto.offset.reset property. If 
auto.offset.reset=latest, we assume a lag of 0. If auto.offset.reset=earliest 
(or any other value) we assume lag equal to the total number of message 
currently available in that partition.

Next, create a map storing the current total lag of all partitions assigned 
to each member of the consumer group. Partitions are assigned in decreasing 
order of lag, with each partition assigned to the consumer with least total 
number of assigned partitions, breaking ties by assigning to the consumer with 
the least total currently assigned lag.

Assigning partitions evenly across consumers (by partition count) ensures 
that the assignment is reasonably balanced (by partition count) when all 
partitions have a current lag of 0 or if the distribution of lags is heavily 
skewed. It also gives the consumer group the best possible chance of remaining 
balanced if the assignment is retained for a long period (assuming throughput 
is consistent across members of the consumer group).

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

$ git pull https://github.com/grantneale/kafka 
feature/kafka-5337-lag-based-assignor

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

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

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

This closes #3158






> Partition assignment strategy that distributes lag evenly across consumers in 
> each group
> 
>
> Key: KAFKA-5337
> URL: https://issues.apache.org/jira/browse/KAFKA-5337
> Project: Kafka
>  Issue Type: New Feature
>  Components: consumer
>Affects Versions: 0.10.2.1
>Reporter: Grant Neale
>Priority: Minor
>
> Existing partition assignment strategies (RangeAssignor and 
> RoundRobinAssignor) do not account for the current consumer group lag on each 
> partition.  This can result in sub-optimal assignments when the distribution 
> of lags for a given topic and consumer group is skewed.
> The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
> partitions such that lag is distributed as evenly across a consumer group.
> h4. Algorithm:
> For each topic, first obtain the lag on all partitions. Lag on a given 
> partition is the difference between the end offset and the last offset 
> committed by the consumer group. If no offsets have been committed for a 
> partition we determine the lag based on the code auto.offset.reset property. 
> If auto.offset.reset=latest, we assume a lag of 0. If 
> auto.offset.reset=earliest (or any other value) we assume lag equal to the 
> total number of message currently available in that partition.
> Next, create a map storing the current total lag of all partitions assigned 
> to each member of the consumer group. Partitions are assigned in decreasing 
> order of lag, with each partition assigned to the consumer with least total 
> number of assigned partitions, breaking ties by assigning to the consumer 
> with the least total currently assigned lag.
> Assigning partitions evenly across consumers (by partition count) ensures 
> that the assignment is reasonably balanced (by partition count) when all 
> partitions have a current lag of 0 or if the distribution of lags is heavily 
> skewed. It also gives the consumer group the best possible chance of 
> remaining balanced if the assignment is retained for a long period (assuming 
> throughput is consistent across members of the consumer group).



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

[GitHub] kafka pull request #3158: KAFKA-5337: LagBasedAssignor partition assignment ...

[grantneale]

GitHub user grantneale opened a pull request:

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

KAFKA-5337: LagBasedAssignor partition assignment strategy

Existing partition assignment strategies (RangeAssignor and 
RoundRobinAssignor) do not account for the current consumer group lag on each 
partition. This can result in sub-optimal assignments when the distribution of 
lags for a given topic and consumer group is skewed.

The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
partitions such that lag is distributed as evenly across a consumer group.

## Algorithm

For each topic, first obtain the lag on all partitions. Lag on a given 
partition is the difference between the end offset and the last offset 
committed by the consumer group. If no offsets have been committed for a 
partition we determine the lag based on the code auto.offset.reset property. If 
auto.offset.reset=latest, we assume a lag of 0. If auto.offset.reset=earliest 
(or any other value) we assume lag equal to the total number of message 
currently available in that partition.

Next, create a map storing the current total lag of all partitions assigned 
to each member of the consumer group. Partitions are assigned in decreasing 
order of lag, with each partition assigned to the consumer with least total 
number of assigned partitions, breaking ties by assigning to the consumer with 
the least total currently assigned lag.

Assigning partitions evenly across consumers (by partition count) ensures 
that the assignment is reasonably balanced (by partition count) when all 
partitions have a current lag of 0 or if the distribution of lags is heavily 
skewed. It also gives the consumer group the best possible chance of remaining 
balanced if the assignment is retained for a long period (assuming throughput 
is consistent across members of the consumer group).

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

$ git pull https://github.com/grantneale/kafka 
feature/kafka-5337-lag-based-assignor

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

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

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

This closes #3158






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

[jira] [Updated] (KAFKA-5319) Add a tool to make cluster replica and leader balance

[Ma Tianchi (JIRA)]

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

Ma Tianchi updated KAFKA-5319:
--
Attachment: (was: AlgorithmDescription.png)

> Add a tool to make cluster replica and leader balance
> -
>
> Key: KAFKA-5319
> URL: https://issues.apache.org/jira/browse/KAFKA-5319
> Project: Kafka
>  Issue Type: New Feature
>  Components: admin
>Affects Versions: 0.10.2.1
>Reporter: Ma Tianchi
>
> When a new broker is added to cluster,there is not any topics in the new 
> broker.When we use console command to create a topic without 
> 'replicaAssignmentOpt',Kafka use AdminUtils.assignReplicasToBrokers to get 
> replicaAssignment.Even though it is balance at the creating time if the 
> cluster never change,with more and more brokers added to cluster the replica 
> balanced will become not well. We also can use 'kafka-reassign-partitions.sh' 
> to balance ,but the big amount of topics make it not easy.And at the topic 
> created time , Kafka choose a  PreferredReplicaLeader which be put at the 
> first position of  the AR to make leader balance.But the balance will be 
> destroyed when cluster changed.Using  'kafka-reassign-partitions.sh' to make 
> partition reassignment may be also destroy the leader balance ,because user 
> can change the AR of the partition . It may be not balanced , but Kafka 
> believe cluster leader balance is well with every leaders is the first on at 
> AR.
> So we create a tool to make the number of replicas and number of leaders on 
> every brokers is balanced.It uses a algorithm to get a balanced replica and 
> leader  reassignment,then uses ReassignPartitionsCommand to real balance the 
> cluster.
> It can be used to make balance when cluster added brokers or cluster is not 
> balanced .It does not deal with moving replicas of a dead broker to an alive 
> broker,it only makes replicas and leaders on alive brokers is balanced.



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

[jira] [Commented] (KAFKA-5337) Partition assignment strategy that distributes lag evenly across consumers in each group

[Grant Neale (JIRA)]

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

Grant Neale commented on KAFKA-5337:


Currently working on this and intend to raise a PR soon.  Unfortunately, I 
don't have permission to assign the issue to myself.

> Partition assignment strategy that distributes lag evenly across consumers in 
> each group
> 
>
> Key: KAFKA-5337
> URL: https://issues.apache.org/jira/browse/KAFKA-5337
> Project: Kafka
>  Issue Type: New Feature
>  Components: consumer
>Affects Versions: 0.10.2.1
>Reporter: Grant Neale
>Priority: Minor
>
> Existing partition assignment strategies (RangeAssignor and 
> RoundRobinAssignor) do not account for the current consumer group lag on each 
> partition.  This can result in sub-optimal assignments when the distribution 
> of lags for a given topic and consumer group is skewed.
> The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
> partitions such that lag is distributed as evenly across a consumer group.
> h4. Algorithm:
> For each topic, first obtain the lag on all partitions. Lag on a given 
> partition is the difference between the end offset and the last offset 
> committed by the consumer group. If no offsets have been committed for a 
> partition we determine the lag based on the code auto.offset.reset property. 
> If auto.offset.reset=latest, we assume a lag of 0. If 
> auto.offset.reset=earliest (or any other value) we assume lag equal to the 
> total number of message currently available in that partition.
> Next, create a map storing the current total lag of all partitions assigned 
> to each member of the consumer group. Partitions are assigned in decreasing 
> order of lag, with each partition assigned to the consumer with least total 
> number of assigned partitions, breaking ties by assigning to the consumer 
> with the least total currently assigned lag.
> Assigning partitions evenly across consumers (by partition count) ensures 
> that the assignment is reasonably balanced (by partition count) when all 
> partitions have a current lag of 0 or if the distribution of lags is heavily 
> skewed. It also gives the consumer group the best possible chance of 
> remaining balanced if the assignment is retained for a long period (assuming 
> throughput is consistent across members of the consumer group).



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

[jira] [Updated] (KAFKA-5337) Partition assignment strategy that distributes lag evenly across consumers in each group

[Grant Neale (JIRA)]

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

Grant Neale updated KAFKA-5337:
---
Description: 
Existing partition assignment strategies (RangeAssignor and RoundRobinAssignor) 
do not account for the current consumer group lag on each partition.  This can 
result in sub-optimal assignments when the distribution of lags for a given 
topic and consumer group is skewed.

The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
partitions such that lag is distributed as evenly across a consumer group.

h4. Algorithm:

For each topic, first obtain the lag on all partitions. Lag on a given 
partition is the difference between the end offset and the last offset 
committed by the consumer group. If no offsets have been committed for a 
partition we determine the lag based on the code auto.offset.reset property. If 
auto.offset.reset=latest, we assume a lag of 0. If auto.offset.reset=earliest 
(or any other value) we assume lag equal to the total number of message 
currently available in that partition.

Next, create a map storing the current total lag of all partitions assigned to 
each member of the consumer group. Partitions are assigned in decreasing order 
of lag, with each partition assigned to the consumer with least total number of 
assigned partitions, breaking ties by assigning to the consumer with the least 
total currently assigned lag.

Assigning partitions evenly across consumers (by partition count) ensures that 
the assignment is reasonably balanced (by partition count) when all partitions 
have a current lag of 0 or if the distribution of lags is heavily skewed. It 
also gives the consumer group the best possible chance of remaining balanced if 
the assignment is retained for a long period (assuming throughput is consistent 
across members of the consumer group).

  was:
Existing partition assignment strategies (Range Assignor and 
RoubdRobinAssignor) do not account for the current consumer group lag on each 
partition.  This can result in sub-optimal assignments when the distribution of 
lags for a given topic and consumer group is skewed.

The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
partitions such that lag is distributed as evenly across a consumer group.

h4. Algorithm:

For each topic, first obtain the lag on all partitions. Lag on a given 
partition is the difference between the end offset and the last offset 
committed by the consumer group. If no offsets have been committed for a 
partition we determine the lag based on the code auto.offset.reset property. If 
auto.offset.reset=latest, we assume a lag of 0. If auto.offset.reset=earliest 
(or any other value) we assume lag equal to the total number of message 
currently available in that partition.

Next, create a map storing the current total lag of all partitions assigned to 
each member of the consumer group. Partitions are assigned in decreasing order 
of lag, with each partition assigned to the consumer with least total number of 
assigned partitions, breaking ties by assigning to the consumer with the least 
total currently assigned lag.

Assigning partitions evenly across consumers (by partition count) ensures that 
the assignment is reasonably balanced (by partition count) when all partitions 
have a current lag of 0 or if the distribution of lags is heavily skewed. It 
also gives the consumer group the best possible chance of remaining balanced if 
the assignment is retained for a long period (assuming throughput is consistent 
across members of the consumer group).


> Partition assignment strategy that distributes lag evenly across consumers in 
> each group
> 
>
> Key: KAFKA-5337
> URL: https://issues.apache.org/jira/browse/KAFKA-5337
> Project: Kafka
>  Issue Type: New Feature
>  Components: consumer
>Affects Versions: 0.10.2.1
>Reporter: Grant Neale
>Priority: Minor
>
> Existing partition assignment strategies (RangeAssignor and 
> RoundRobinAssignor) do not account for the current consumer group lag on each 
> partition.  This can result in sub-optimal assignments when the distribution 
> of lags for a given topic and consumer group is skewed.
> The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
> partitions such that lag is distributed as evenly across a consumer group.
> h4. Algorithm:
> For each topic, first obtain the lag on all partitions. Lag on a given 
> partition is the difference between the end offset and the last offset 
> committed by the consumer group. If no offsets have been committed for a 
> partition we determine the lag based on the code auto.offset.reset property. 
> If auto.offset.reset=latest, we assume a lag of 0. If 
> auto.offset.reset=earlie

[jira] [Updated] (KAFKA-5337) Partition assignment strategy that distributes lag evenly across consumers in each group

[Grant Neale (JIRA)]

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

Grant Neale updated KAFKA-5337:
---
Description: 
Existing partition assignment strategies (Range Assignor and 
RoubdRobinAssignor) do not account for the current consumer group lag on each 
partition.  This can result in sub-optimal assignments when the distribution of 
lags for a given topic and consumer group is skewed.

The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
partitions such that lag is distributed as evenly across a consumer group.

h4. Algorithm:

For each topic, first obtain the lag on all partitions. Lag on a given 
partition is the difference between the end offset and the last offset 
committed by the consumer group. If no offsets have been committed for a 
partition we determine the lag based on the code auto.offset.reset property. If 
auto.offset.reset=latest, we assume a lag of 0. If auto.offset.reset=earliest 
(or any other value) we assume lag equal to the total number of message 
currently available in that partition.

Next, create a map storing the current total lag of all partitions assigned to 
each member of the consumer group. Partitions are assigned in decreasing order 
of lag, with each partition assigned to the consumer with least total number of 
assigned partitions, breaking ties by assigning to the consumer with the least 
total currently assigned lag.

Assigning partitions evenly across consumers (by partition count) ensures that 
the assignment is reasonably balanced (by partition count) when all partitions 
have a current lag of 0 or if the distribution of lags is heavily skewed. It 
also gives the consumer group the best possible chance of remaining balanced if 
the assignment is retained for a long period (assuming throughput is consistent 
across members of the consumer group).

  was:
Existing partition assignment strategies (Range Assignor and 
RoubdRobinAssignor) do not account for the current consumer group lag on each 
partition.  This can result in sub-optimal assignments when the distribution of 
lags for a given topic and consumer group is skewed.

The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
partitions such that lag is distributed as evenly across a consumer group.

**Algorithm:**

For each topic, first obtain the lag on all partitions. Lag on a given 
partition is the difference between the end offset and the last offset 
committed by the consumer group. If no offsets have been committed for a 
partition we determine the lag based on the code auto.offset.reset property. If 
auto.offset.reset=latest, we assume a lag of 0. If auto.offset.reset=earliest 
(or any other value) we assume lag equal to the total number of message 
currently available in that partition.

Next, create a map storing the current total lag of all partitions assigned to 
each member of the consumer group. Partitions are assigned in decreasing order 
of lag, with each partition assigned to the consumer with least total number of 
assigned partitions, breaking ties by assigning to the consumer with the least 
total currently assigned lag.

Assigning partitions evenly across consumers (by partition count) ensures that 
the assignment is reasonably balanced (by partition count) when all partitions 
have a current lag of 0 or if the distribution of lags is heavily skewed. It 
also gives the consumer group the best possible chance of remaining balanced if 
the assignment is retained for a long period (assuming throughput is consistent 
across members of the consumer group).


> Partition assignment strategy that distributes lag evenly across consumers in 
> each group
> 
>
> Key: KAFKA-5337
> URL: https://issues.apache.org/jira/browse/KAFKA-5337
> Project: Kafka
>  Issue Type: New Feature
>  Components: consumer
>Affects Versions: 0.10.2.1
>Reporter: Grant Neale
>Priority: Minor
>
> Existing partition assignment strategies (Range Assignor and 
> RoubdRobinAssignor) do not account for the current consumer group lag on each 
> partition.  This can result in sub-optimal assignments when the distribution 
> of lags for a given topic and consumer group is skewed.
> The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
> partitions such that lag is distributed as evenly across a consumer group.
> h4. Algorithm:
> For each topic, first obtain the lag on all partitions. Lag on a given 
> partition is the difference between the end offset and the last offset 
> committed by the consumer group. If no offsets have been committed for a 
> partition we determine the lag based on the code auto.offset.reset property. 
> If auto.offset.reset=latest, we assume a lag of 0. If 
> auto.offset.reset=earl

[jira] [Created] (KAFKA-5337) Partition assignment strategy that distributes lag evenly across consumers in each group

[Grant Neale (JIRA)]

Grant Neale created KAFKA-5337:
--

 Summary: Partition assignment strategy that distributes lag evenly 
across consumers in each group
 Key: KAFKA-5337
 URL: https://issues.apache.org/jira/browse/KAFKA-5337
 Project: Kafka
  Issue Type: New Feature
  Components: consumer
Affects Versions: 0.10.2.1
Reporter: Grant Neale
Priority: Minor


Existing partition assignment strategies (Range Assignor and 
RoubdRobinAssignor) do not account for the current consumer group lag on each 
partition.  This can result in sub-optimal assignments when the distribution of 
lags for a given topic and consumer group is skewed.

The LagBasedAssignor operates on a per-topic basis, and attempts to assign 
partitions such that lag is distributed as evenly across a consumer group.

**Algorithm:**

For each topic, first obtain the lag on all partitions. Lag on a given 
partition is the difference between the end offset and the last offset 
committed by the consumer group. If no offsets have been committed for a 
partition we determine the lag based on the code auto.offset.reset property. If 
auto.offset.reset=latest, we assume a lag of 0. If auto.offset.reset=earliest 
(or any other value) we assume lag equal to the total number of message 
currently available in that partition.

Next, create a map storing the current total lag of all partitions assigned to 
each member of the consumer group. Partitions are assigned in decreasing order 
of lag, with each partition assigned to the consumer with least total number of 
assigned partitions, breaking ties by assigning to the consumer with the least 
total currently assigned lag.

Assigning partitions evenly across consumers (by partition count) ensures that 
the assignment is reasonably balanced (by partition count) when all partitions 
have a current lag of 0 or if the distribution of lags is heavily skewed. It 
also gives the consumer group the best possible chance of remaining balanced if 
the assignment is retained for a long period (assuming throughput is consistent 
across members of the consumer group).



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

[jira] [Assigned] (KAFKA-5171) TC should not accept empty string transactional id

[Umesh Chaudhary (JIRA)]

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

Umesh Chaudhary reassigned KAFKA-5171:
--

Assignee: Umesh Chaudhary

> TC should not accept empty string transactional id
> --
>
> Key: KAFKA-5171
> URL: https://issues.apache.org/jira/browse/KAFKA-5171
> Project: Kafka
>  Issue Type: Sub-task
>  Components: core
>Reporter: Guozhang Wang
>Assignee: Umesh Chaudhary
> Fix For: 0.11.0.0
>
>
> Currently on TC, both `null` and `empty string` will be accepted and a new 
> pid will be returned. However, on the producer client end empty string 
> transactional id is not allowed, and if user specifically set it with empty 
> string RTE will be thrown.
> We can make TC's behavior consistent with client to also reject empty string 
> transactional id.



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

[jira] [Commented] (KAFKA-5171) TC should not accept empty string transactional id

[ASF GitHub Bot (JIRA)]

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

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

Github user umesh9794 closed the pull request at:

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


> TC should not accept empty string transactional id
> --
>
> Key: KAFKA-5171
> URL: https://issues.apache.org/jira/browse/KAFKA-5171
> Project: Kafka
>  Issue Type: Sub-task
>  Components: core
>Reporter: Guozhang Wang
> Fix For: 0.11.0.0
>
>
> Currently on TC, both `null` and `empty string` will be accepted and a new 
> pid will be returned. However, on the producer client end empty string 
> transactional id is not allowed, and if user specifically set it with empty 
> string RTE will be thrown.
> We can make TC's behavior consistent with client to also reject empty string 
> transactional id.



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

[GitHub] kafka pull request #2973: KAFKA-5171 : TC should not accept empty string tra...

[umesh9794]

Github user umesh9794 closed the pull request at:

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


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

[jira] [Commented] (KAFKA-5332) When resize the index file, maybe caused the content disappear

[xuzq (JIRA)]

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

xuzq commented on KAFKA-5332:
-

@ [~guozhang] Can you check this bug? Thank you.

> When resize the index file, maybe caused the content disappear
> --
>
> Key: KAFKA-5332
> URL: https://issues.apache.org/jira/browse/KAFKA-5332
> Project: Kafka
>  Issue Type: Bug
>  Components: core, log
>Affects Versions: 0.10.2.1
>Reporter: xuzq
>
> When resize the index file, maybe caused the content disappear.
> When the kafka server is running, someone removed the index file on the disk 
> manually, if at this point, the function AbstractIndex.Resize(newSize: Int) 
> is triggered, it will create a new .index file which the size is 
> roundedNewSize, but the content is empty. 
> After this, the contents of mmap is empty. 
> When looking for specific offset corresponds to position, it also return 
> zero. The index file to locate the position does not provide any convenience.
> I think if the ".index" is not exist, we should copy the contents from old 
> "mmap" to new "mmap" to avoid the "empty file".



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

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

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-4660; Improve test coverage KafkaStreams

[wangguoz] KAFKA-5191: Autogenerate Consumer Fetcher metrics

--
[...truncated 2.95 MB...]
kafka.security.auth.SimpleAclAuthorizerTest > testDenyTakesPrecedence PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testNoAclFoundOverride STARTED

kafka.security.auth.SimpleAclAuthorizerTest > testNoAclFoundOverride PASSED

kafka.security.auth.SimpleAclAuthorizerTest > 
testHighConcurrencyModificationOfResourceAcls STARTED

kafka.security.auth.SimpleAclAuthorizerTest > 
testHighConcurrencyModificationOfResourceAcls PASSED

kafka.security.auth.SimpleAclAuthorizerTest > testLoadCache STARTED

kafka.security.auth.SimpleAclAuthorizerTest > testLoadCache PASSED

kafka.integration.PrimitiveApiTest > testMultiProduce STARTED

kafka.integration.PrimitiveApiTest > testMultiProduce PASSED

kafka.integration.PrimitiveApiTest > testDefaultEncoderProducerAndFetch STARTED

kafka.integration.PrimitiveApiTest > testDefaultEncoderProducerAndFetch PASSED

kafka.integration.PrimitiveApiTest > testFetchRequestCanProperlySerialize 
STARTED

kafka.integration.PrimitiveApiTest > testFetchRequestCanProperlySerialize PASSED

kafka.integration.PrimitiveApiTest > testPipelinedProduceRequests STARTED

kafka.integration.PrimitiveApiTest > testPipelinedProduceRequests PASSED

kafka.integration.PrimitiveApiTest > testProduceAndMultiFetch STARTED

kafka.integration.PrimitiveApiTest > testProduceAndMultiFetch PASSED

kafka.integration.PrimitiveApiTest > 
testDefaultEncoderProducerAndFetchWithCompression STARTED

kafka.integration.PrimitiveApiTest > 
testDefaultEncoderProducerAndFetchWithCompression PASSED

kafka.integration.PrimitiveApiTest > testConsumerEmptyTopic STARTED

kafka.integration.PrimitiveApiTest > testConsumerEmptyTopic PASSED

kafka.integration.PrimitiveApiTest > testEmptyFetchRequest STARTED

kafka.integration.PrimitiveApiTest > testEmptyFetchRequest PASSED

kafka.integration.MetricsDuringTopicCreationDeletionTest > 
testMetricsDuringTopicCreateDelete STARTED

kafka.integration.MetricsDuringTopicCreationDeletionTest > 
testMetricsDuringTopicCreateDelete PASSED

kafka.integration.AutoOffsetResetTest > testResetToLatestWhenOffsetTooLow 
STARTED

kafka.integration.AutoOffsetResetTest > testResetToLatestWhenOffsetTooLow PASSED

kafka.integration.AutoOffsetResetTest > testResetToEarliestWhenOffsetTooLow 
STARTED

kafka.integration.AutoOffsetResetTest > testResetToEarliestWhenOffsetTooLow 
PASSED

kafka.integration.AutoOffsetResetTest > testResetToEarliestWhenOffsetTooHigh 
STARTED

kafka.integration.AutoOffsetResetTest > testResetToEarliestWhenOffsetTooHigh 
PASSED

kafka.integration.AutoOffsetResetTest > testResetToLatestWhenOffsetTooHigh 
STARTED

kafka.integration.AutoOffsetResetTest > testResetToLatestWhenOffsetTooHigh 
PASSED

kafka.integration.TopicMetadataTest > testIsrAfterBrokerShutDownAndJoinsBack 
STARTED

kafka.integration.TopicMetadataTest > testIsrAfterBrokerShutDownAndJoinsBack 
PASSED

kafka.integration.TopicMetadataTest > testAutoCreateTopicWithCollision STARTED

kafka.integration.TopicMetadataTest > testAutoCreateTopicWithCollision PASSED

kafka.integration.TopicMetadataTest > testAliveBrokerListWithNoTopics STARTED

kafka.integration.TopicMetadataTest > testAliveBrokerListWithNoTopics PASSED

kafka.integration.TopicMetadataTest > testAutoCreateTopic STARTED

kafka.integration.TopicMetadataTest > testAutoCreateTopic PASSED

kafka.integration.TopicMetadataTest > testGetAllTopicMetadata STARTED

kafka.integration.TopicMetadataTest > testGetAllTopicMetadata PASSED

kafka.integration.TopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup STARTED

kafka.integration.TopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterNewBrokerStartup PASSED

kafka.integration.TopicMetadataTest > testBasicTopicMetadata STARTED

kafka.integration.TopicMetadataTest > testBasicTopicMetadata PASSED

kafka.integration.TopicMetadataTest > testAutoCreateTopicWithInvalidReplication 
STARTED

kafka.integration.TopicMetadataTest > testAutoCreateTopicWithInvalidReplication 
PASSED

kafka.integration.TopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown STARTED

kafka.integration.TopicMetadataTest > 
testAliveBrokersListWithNoTopicsAfterABrokerShutdown PASSED

kafka.integration.FetcherTest > testFetcher STARTED

kafka.integration.FetcherTest > testFetcher PASSED

kafka.integration.UncleanLeaderElectionTest > testUncleanLeaderElectionEnabled 
STARTED

kafka.integration.UncleanLeaderElectionTest > testUncleanLeaderElectionEnabled 
PASSED

kafka.integration.UncleanLeaderElectionTest > 
testCleanLeaderElectionDisabledByTopicOverride STARTED

kafka.integration.UncleanLeaderElectionTest > 
testCleanLeaderElectionDisabledByTopicOverride SKIPPED

kafka.integration.UncleanLeaderElectionTest > testUncleanLeaderElectio

Re: KIP-162: Enable topic deletion by default

[Onur Karaman]

Would it make sense to resolve KAFKA-4893 before enabling it by default, as
fixing the ticket would likely involve changing the log directory structure?

On Fri, May 26, 2017 at 3:24 PM, Guozhang Wang  wrote:

> I'd say just remove those two lines.
>
> On Fri, May 26, 2017 at 7:55 AM, Gwen Shapira  wrote:
>
> > This was a discussion, not a vote (sorry for mangling the title), but
> > thanks for all the +1 anyway.
> >
> > Regarding Ismael's feedback:
> > The current server.properties includes the following:
> > # Switch to enable topic deletion or not, default value is false
> > #delete.topic.enable=true
> >
> > We can't leave it as is, obviously - since the KIP invalidates the
> > comment.  Lets just remove those two lines?
> >
> > Note that all our proposed changes may break few community puppet/docker
> > scripts that use these lines for "sed" that enables topic deletion.
> >
> > Gwen
> >
> > On Fri, May 26, 2017 at 5:41 PM Tom Crayford 
> wrote:
> >
> > > +1 (non-binding)
> > >
> > > On Fri, May 26, 2017 at 3:38 PM, Damian Guy 
> > wrote:
> > >
> > > > +1
> > > > Also agree with what Ismael said.
> > > >
> > > > On Fri, 26 May 2017 at 15:26 Ismael Juma  wrote:
> > > >
> > > > > Thanks for the KIP, sounds good to me. One comment: not sure we
> need
> > to
> > > > add
> > > > > the config to server.properties. Do we expect people to change this
> > > > > default?
> > > > >
> > > > > On Fri, May 26, 2017 at 3:03 PM, Gwen Shapira 
> > > wrote:
> > > > >
> > > > > > Hi Kafka developers, users and friends,
> > > > > >
> > > > > > I've added a KIP to improve our out-of-the-box usability a bit:
> > > > > > KIP-162: Enable topic deletion by default:
> > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+
> > > > > > Enable+topic+deletion+by+default
> > > > > >
> > > > > > Pretty simple :) Discussion and feedback are welcome.
> > > > > >
> > > > > > Gwen
> > > > > >
> > > > >
> > > >
> > >
> >
>
>
>
> --
> -- Guozhang
>

[jira] [Commented] (KAFKA-5328) consider switching json parser from scala to jackson

[Onur Karaman (JIRA)]

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

Onur Karaman commented on KAFKA-5328:
-

I copied the topic assignment znodes and partition state znodes from a cluster 
into two text files and just timed how long it took to parse the topic and 
partition znodes using both the scala json parser and jackson.

This experiment had 3066 topics and 95895 partitions (the actual cluster I got 
the znodes from was actually a third the size but I just duplicated lines to 
reflect some of our other clusters).

Here's the code:
{code}
/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package kafka.tools

import java.nio.file.{Files, Paths}

import com.fasterxml.jackson.databind.ObjectMapper
import kafka.api.LeaderAndIsr
import kafka.common.TopicAndPartition
import kafka.controller.LeaderIsrAndControllerEpoch
import kafka.utils.Json

import scala.collection.JavaConverters._
import scala.collection.Seq

object JsonPerformance {
  def main(args: Array[String]): Unit = {
val assignments = 
Files.readAllLines(Paths.get("/Users/okaraman/code/read-zk-state/raw_assignments_big.txt"))
val states = 
Files.readAllLines(Paths.get("/Users/okaraman/code/read-zk-state/raw_states_big.txt"))

time(assignments.asScala.foreach(parseAssignmentWithScala), "scala json 
parser assignments")
time(assignments.asScala.foreach(parseAssignmentWithJackson), "jackson json 
parser assignments")
time(states.asScala.foreach(parseStateWithScala), "scala json parser 
states")
time(states.asScala.foreach(parseStateWithJackson), "jackson json parser 
states")
  }

  def time(f: => Unit, messagePrefix: String): Unit = {
val start = System.currentTimeMillis()
f
val end = System.currentTimeMillis()
println(messagePrefix + s" took: ${end - start} ms")
  }

  def parseAssignmentWithScala(assignment: String): Map[TopicAndPartition, 
Seq[Int]] = {
val json = Json.parseFull(assignment).get
val jsonPartitions = json.asInstanceOf[Map[String, 
Any]].get("partitions").get
val replicaMap = jsonPartitions.asInstanceOf[Map[String, Seq[Int]]]
replicaMap.map { case (partition, replicas) => TopicAndPartition("t", 
partition.toInt) -> replicas }
  }

  def parseAssignmentWithJackson(assignment: String): Map[TopicAndPartition, 
Seq[Int]] = {
val mapper = new ObjectMapper()
val json = mapper.readTree(assignment)
json.get("partitions").asScala.toList.flatMap(x => (0 until x.size()).map(i 
=> TopicAndPartition("t", i) -> x.get(i).asScala.toSeq.map(_.asInt(.toMap
  }

  def parseStateWithScala(state: String): LeaderIsrAndControllerEpoch = {
val json = Json.parseFull(state).get
val leaderIsrAndEpochInfo = json.asInstanceOf[Map[String, Any]]
val leader = leaderIsrAndEpochInfo.get("leader").get.asInstanceOf[Int]
val epoch = leaderIsrAndEpochInfo.get("leader_epoch").get.asInstanceOf[Int]
val isr = leaderIsrAndEpochInfo.get("isr").get.asInstanceOf[List[Int]]
val controllerEpoch = 
leaderIsrAndEpochInfo.get("controller_epoch").get.asInstanceOf[Int]
LeaderIsrAndControllerEpoch(LeaderAndIsr(leader, epoch, isr, 1), 
controllerEpoch)
  }

  def parseStateWithJackson(state: String): LeaderIsrAndControllerEpoch = {
val mapper = new ObjectMapper()
val json = mapper.readTree(state)
val leader = json.get("leader").asInt()
val epoch = json.get("leader_epoch").asInt()
val isr = json.get("isr").asScala.toList.map(_.asInt())
val controllerEpoch = json.get("controller_epoch").asInt()
LeaderIsrAndControllerEpoch(LeaderAndIsr(leader, epoch, isr, 1), 
controllerEpoch)
  }
}
{code}

Here's the output:
{code}
scala json parser assignments took: 8730 ms
jackson json parser assignments took: 354 ms
scala json parser states took: 28395 ms
jackson json parser states took: 360 ms
{code}

So controller initialization time spent on json parsing would be reduced from 
37.1 seconds down to 0.7 seconds.

> consider switching json parser from scala to jackson
> 
>
> Key: KAFKA-5328
> URL: https://issues.apa

[jira] [Created] (KAFKA-5336) The required ACL permission for ListGroup is invalid

[Vahid Hashemian (JIRA)]

Vahid Hashemian created KAFKA-5336:
--

 Summary: The required ACL permission for ListGroup is invalid
 Key: KAFKA-5336
 URL: https://issues.apache.org/jira/browse/KAFKA-5336
 Project: Kafka
  Issue Type: Bug
  Components: security
Affects Versions: 0.10.2.1
Reporter: Vahid Hashemian
Assignee: Vahid Hashemian
Priority: Minor


The {{ListGroup}} API authorizes requests with _Describe_ access to the cluster 
resource:

{code}
  def handleListGroupsRequest(request: RequestChannel.Request) {
if (!authorize(request.session, Describe, Resource.ClusterResource)) {
  sendResponseMaybeThrottle(request, requestThrottleMs =>
ListGroupsResponse.fromError(requestThrottleMs, 
Errors.CLUSTER_AUTHORIZATION_FAILED))
} else {
  ...
{code}

 However, the list of operations (or permissions) allowed for the cluster 
resource does not include _Describe_:
{code}
  val ResourceTypeToValidOperations = Map[ResourceType, Set[Operation]] (
...
Cluster -> Set(Create, ClusterAction, DescribeConfigs, AlterConfigs, 
IdempotentWrite, All),
...
  )
{code}

Only a user with _All_ cluster permission can successfully call the 
{{ListGroup}} API. No other permission (not even any combination that does not 
include _All_) would let user use this API.

The bug could be as simple as a typo in the API handler. Though it's not 
obvious what actual permission was meant to be used there (perhaps 
_DescribeConfigs_?)



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

[jira] [Commented] (KAFKA-5211) KafkaConsumer should not skip a corrupted record after throwing an exception.

[Guozhang Wang (JIRA)]

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

Guozhang Wang commented on KAFKA-5211:
--

For this specific issue, Streams should be independent of whichever behavior in 
Consumer since Streams always use the {{Consumer}} and perform 
deserialization after polling the records. This is because embedded consumer 
within Streams would probably fetch from topics with different serde 
mechanisms, so it need to deserialize based on the topics.

> KafkaConsumer should not skip a corrupted record after throwing an exception.
> -
>
> Key: KAFKA-5211
> URL: https://issues.apache.org/jira/browse/KAFKA-5211
> Project: Kafka
>  Issue Type: Bug
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
>  Labels: clients, consumer
> Fix For: 0.11.0.0
>
>
> In 0.10.2, when there is a corrupted record, KafkaConsumer.poll() will throw 
> an exception and block on that corrupted record. In the latest trunk this 
> behavior has changed to skip the corrupted record (which is the old consumer 
> behavior). With KIP-98, skipping corrupted messages would be a little 
> dangerous as the message could be a control message for a transaction. We 
> should fix the issue to let the KafkaConsumer block on the corrupted messages.



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

[jira] [Commented] (KAFKA-5191) Autogenerate Consumer Fetcher metrics

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Autogenerate Consumer Fetcher metrics
> -
>
> Key: KAFKA-5191
> URL: https://issues.apache.org/jira/browse/KAFKA-5191
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: James Cheng
>Assignee: James Cheng
> Fix For: 0.11.0.0
>
> Attachments: docs_now_include_partition_level_metrics.png, 
> generated_fetcher_docs.png, generated_fetcher_docs_with_alternate_css.png, 
> generated_fetcher_docs_with_css.png
>
>




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

[GitHub] kafka pull request #2993: KAFKA-5191: Autogenerate Consumer Fetcher metrics

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Updated] (KAFKA-5191) Autogenerate Consumer Fetcher metrics

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-5191:
-
   Resolution: Fixed
Fix Version/s: 0.11.0.0
   Status: Resolved  (was: Patch Available)

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

> Autogenerate Consumer Fetcher metrics
> -
>
> Key: KAFKA-5191
> URL: https://issues.apache.org/jira/browse/KAFKA-5191
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: James Cheng
>Assignee: James Cheng
> Fix For: 0.11.0.0
>
> Attachments: docs_now_include_partition_level_metrics.png, 
> generated_fetcher_docs.png, generated_fetcher_docs_with_alternate_css.png, 
> generated_fetcher_docs_with_css.png
>
>




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

[jira] [Commented] (KAFKA-4660) Improve test coverage KafkaStreams

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Improve test coverage KafkaStreams
> --
>
> Key: KAFKA-4660
> URL: https://issues.apache.org/jira/browse/KAFKA-4660
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Umesh Chaudhary
>Priority: Minor
> Fix For: 0.11.0.0
>
>
> {{toString}} is used to print the topology, so probably should have a unit 
> test.



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

[GitHub] kafka pull request #3099: KAFKA-4660 : Improve test coverage KafkaStreams

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Updated] (KAFKA-4660) Improve test coverage KafkaStreams

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-4660:
-
Resolution: Fixed
Status: Resolved  (was: Patch Available)

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

> Improve test coverage KafkaStreams
> --
>
> Key: KAFKA-4660
> URL: https://issues.apache.org/jira/browse/KAFKA-4660
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Umesh Chaudhary
>Priority: Minor
> Fix For: 0.11.0.0
>
>
> {{toString}} is used to print the topology, so probably should have a unit 
> test.



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

[jira] [Commented] (KAFKA-5333) Remove Broker ACL resource type

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Remove Broker ACL resource type
> ---
>
> Key: KAFKA-5333
> URL: https://issues.apache.org/jira/browse/KAFKA-5333
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ismael Juma
>Assignee: Ismael Juma
>Priority: Critical
> Fix For: 0.11.0.0
>
>
> We added it as part of KIP-133 in the context of describing and altering 
> configs, but Jun pointed out that it does not give us much over the Cluster 
> resource type. We can always add it later, if necessary.



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

[GitHub] kafka pull request #3154: KAFKA-5333; Remove Broker ACL resource type

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Updated] (KAFKA-5333) Remove Broker ACL resource type

[Jun Rao (JIRA)]

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

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

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

> Remove Broker ACL resource type
> ---
>
> Key: KAFKA-5333
> URL: https://issues.apache.org/jira/browse/KAFKA-5333
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ismael Juma
>Assignee: Ismael Juma
>Priority: Critical
> Fix For: 0.11.0.0
>
>
> We added it as part of KIP-133 in the context of describing and altering 
> configs, but Jun pointed out that it does not give us much over the Cluster 
> resource type. We can always add it later, if necessary.



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

Re: KIP-162: Enable topic deletion by default

[Guozhang Wang]

I'd say just remove those two lines.

On Fri, May 26, 2017 at 7:55 AM, Gwen Shapira  wrote:

> This was a discussion, not a vote (sorry for mangling the title), but
> thanks for all the +1 anyway.
>
> Regarding Ismael's feedback:
> The current server.properties includes the following:
> # Switch to enable topic deletion or not, default value is false
> #delete.topic.enable=true
>
> We can't leave it as is, obviously - since the KIP invalidates the
> comment.  Lets just remove those two lines?
>
> Note that all our proposed changes may break few community puppet/docker
> scripts that use these lines for "sed" that enables topic deletion.
>
> Gwen
>
> On Fri, May 26, 2017 at 5:41 PM Tom Crayford  wrote:
>
> > +1 (non-binding)
> >
> > On Fri, May 26, 2017 at 3:38 PM, Damian Guy 
> wrote:
> >
> > > +1
> > > Also agree with what Ismael said.
> > >
> > > On Fri, 26 May 2017 at 15:26 Ismael Juma  wrote:
> > >
> > > > Thanks for the KIP, sounds good to me. One comment: not sure we need
> to
> > > add
> > > > the config to server.properties. Do we expect people to change this
> > > > default?
> > > >
> > > > On Fri, May 26, 2017 at 3:03 PM, Gwen Shapira 
> > wrote:
> > > >
> > > > > Hi Kafka developers, users and friends,
> > > > >
> > > > > I've added a KIP to improve our out-of-the-box usability a bit:
> > > > > KIP-162: Enable topic deletion by default:
> > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+
> > > > > Enable+topic+deletion+by+default
> > > > >
> > > > > Pretty simple :) Discussion and feedback are welcome.
> > > > >
> > > > > Gwen
> > > > >
> > > >
> > >
> >
>



-- 
-- Guozhang

[jira] [Updated] (KAFKA-5226) NullPointerException (NPE) in SourceNodeRecordDeserializer.deserialize

[Bill Bejeck (JIRA)]

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

Bill Bejeck updated KAFKA-5226:
---
Status: Patch Available  (was: In Progress)

> NullPointerException (NPE) in SourceNodeRecordDeserializer.deserialize
> --
>
> Key: KAFKA-5226
> URL: https://issues.apache.org/jira/browse/KAFKA-5226
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.1
> Environment: 64-bit Amazon Linux, JDK8
>Reporter: Ian Springer
>Assignee: Bill Bejeck
> Fix For: 0.11.0.0
>
> Attachments: kafka.log, streamsAppOne.log, streamsAppThree.log, 
> streamsAppTwo.log
>
>
> I saw the following NPE in our Kafka Streams app, which has 3 nodes running 
> on 3 separate machines.. Out of hundreds of messages processed, the NPE only 
> occurred twice. I are not sure of the cause, so I am unable to reproduce it. 
> I'm hoping the Kafka Streams team can guess the cause based on the stack 
> trace. If I can provide any additional details about our app, please let me 
> know.
>  
> {code}
> INFO  2017-05-10 02:58:26,021 org.apache.kafka.common.utils.AppInfoParser  
> Kafka version : 0.10.2.1
> INFO  2017-05-10 02:58:26,021 org.apache.kafka.common.utils.AppInfoParser  
> Kafka commitId : e89bffd6b2eff799
> INFO  2017-05-10 02:58:26,031 o.s.context.support.DefaultLifecycleProcessor  
> Starting beans in phase 0
> INFO  2017-05-10 02:58:26,075 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from CREATED to RUNNING.
> INFO  2017-05-10 02:58:26,075 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] Started 
> Kafka Stream process
> INFO  2017-05-10 02:58:26,086 o.a.k.c.consumer.internals.AbstractCoordinator  
> Discovered coordinator p1kaf1.prod.apptegic.com:9092 (id: 2147482646 rack: 
> null) for group evergage-app.
> INFO  2017-05-10 02:58:26,126 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partitions [] for group evergage-app
> INFO  2017-05-10 02:58:26,126 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from RUNNING to REBALANCING.
> INFO  2017-05-10 02:58:26,127 o.a.k.c.consumer.internals.AbstractCoordinator  
> (Re-)joining group evergage-app
> INFO  2017-05-10 02:58:27,712 o.a.k.c.consumer.internals.AbstractCoordinator  
> Successfully joined group evergage-app with generation 18
> INFO  2017-05-10 02:58:27,716 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Setting newly assigned partitions [us.app.Trigger-0] for group evergage-app
> INFO  2017-05-10 02:58:27,716 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to REBALANCING.
> INFO  2017-05-10 02:58:27,729 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> state stores
> INFO  2017-05-10 02:58:27,731 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> processor nodes of the topology
> INFO  2017-05-10 02:58:27,742 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to RUNNING.
> [14 hours pass...]
> INFO  2017-05-10 16:21:27,476 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partitions [us.app.Trigger-0] for group 
> evergage-app
> INFO  2017-05-10 16:21:27,477 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from RUNNING to REBALANCING.
> INFO  2017-05-10 16:21:27,482 o.a.k.c.consumer.internals.AbstractCoordinator  
> (Re-)joining group evergage-app
> INFO  2017-05-10 16:21:27,489 o.a.k.c.consumer.internals.AbstractCoordinator  
> Successfully joined group evergage-app with generation 19
> INFO  2017-05-10 16:21:27,489 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Setting newly assigned partitions [us.app.Trigger-0] for group evergage-app
> INFO  2017-05-10 16:21:27,489 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to REBALANCING.
> INFO  2017-05-10 16:21:27,489 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> processor nodes of the topology
> INFO  2017-05-10 16:21:27,493 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to RUNNING.
> INFO  2017-05-10 16:21:30,584 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partiti

[jira] [Commented] (KAFKA-5226) NullPointerException (NPE) in SourceNodeRecordDeserializer.deserialize

[ASF GitHub Bot (JIRA)]

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

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

GitHub user bbejeck opened a pull request:

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

KAFKA-5226: Fixes issue where adding topics matching a regex

subscribed stream may not be detected by all followers until
onJoinComplete returns.

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

$ git pull https://github.com/bbejeck/kafka 
KAFKA-5226_null_pointer_source_node_deserialize

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

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

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

This closes #3157


commit ed3c3602f7ac794c29686e95ade822dd22d7be97
Author: Bill Bejeck 
Date:   2017-05-26T21:42:14Z

KAFKA-5226: Fixes issue where adding topics matching a regex
subscribed stream may not be detected by all followers until
onJoinComplete returns.




> NullPointerException (NPE) in SourceNodeRecordDeserializer.deserialize
> --
>
> Key: KAFKA-5226
> URL: https://issues.apache.org/jira/browse/KAFKA-5226
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.1
> Environment: 64-bit Amazon Linux, JDK8
>Reporter: Ian Springer
>Assignee: Bill Bejeck
> Fix For: 0.11.0.0
>
> Attachments: kafka.log, streamsAppOne.log, streamsAppThree.log, 
> streamsAppTwo.log
>
>
> I saw the following NPE in our Kafka Streams app, which has 3 nodes running 
> on 3 separate machines.. Out of hundreds of messages processed, the NPE only 
> occurred twice. I are not sure of the cause, so I am unable to reproduce it. 
> I'm hoping the Kafka Streams team can guess the cause based on the stack 
> trace. If I can provide any additional details about our app, please let me 
> know.
>  
> {code}
> INFO  2017-05-10 02:58:26,021 org.apache.kafka.common.utils.AppInfoParser  
> Kafka version : 0.10.2.1
> INFO  2017-05-10 02:58:26,021 org.apache.kafka.common.utils.AppInfoParser  
> Kafka commitId : e89bffd6b2eff799
> INFO  2017-05-10 02:58:26,031 o.s.context.support.DefaultLifecycleProcessor  
> Starting beans in phase 0
> INFO  2017-05-10 02:58:26,075 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from CREATED to RUNNING.
> INFO  2017-05-10 02:58:26,075 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] Started 
> Kafka Stream process
> INFO  2017-05-10 02:58:26,086 o.a.k.c.consumer.internals.AbstractCoordinator  
> Discovered coordinator p1kaf1.prod.apptegic.com:9092 (id: 2147482646 rack: 
> null) for group evergage-app.
> INFO  2017-05-10 02:58:26,126 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partitions [] for group evergage-app
> INFO  2017-05-10 02:58:26,126 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from RUNNING to REBALANCING.
> INFO  2017-05-10 02:58:26,127 o.a.k.c.consumer.internals.AbstractCoordinator  
> (Re-)joining group evergage-app
> INFO  2017-05-10 02:58:27,712 o.a.k.c.consumer.internals.AbstractCoordinator  
> Successfully joined group evergage-app with generation 18
> INFO  2017-05-10 02:58:27,716 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Setting newly assigned partitions [us.app.Trigger-0] for group evergage-app
> INFO  2017-05-10 02:58:27,716 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to REBALANCING.
> INFO  2017-05-10 02:58:27,729 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> state stores
> INFO  2017-05-10 02:58:27,731 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> processor nodes of the topology
> INFO  2017-05-10 02:58:27,742 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to RUNNING.
> [14 hours pass...]
> INFO  2017-05-10 16:21:27,476 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partitions [us.app.Trigger-0] for group 
> evergage-app
> INFO  2017-05-10 16:21:27,477 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from RUNNING to REBALANCING.
> INFO  2017-05-10 16:21:27,482 o.a.k.c.consumer.internals.AbstractCoordinator  
> (

[GitHub] kafka pull request #3157: KAFKA-5226: Fixes issue where adding topics matchi...

[bbejeck]

GitHub user bbejeck opened a pull request:

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

KAFKA-5226: Fixes issue where adding topics matching a regex

subscribed stream may not be detected by all followers until
onJoinComplete returns.

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

$ git pull https://github.com/bbejeck/kafka 
KAFKA-5226_null_pointer_source_node_deserialize

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

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

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

This closes #3157


commit ed3c3602f7ac794c29686e95ade822dd22d7be97
Author: Bill Bejeck 
Date:   2017-05-26T21:42:14Z

KAFKA-5226: Fixes issue where adding topics matching a regex
subscribed stream may not be detected by all followers until
onJoinComplete returns.




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

Re: KIP-162: Enable topic deletion by default

[Jim Jagielski]

> On May 26, 2017, at 1:10 PM, Vahid S Hashemian  
> wrote:
> 
> Gwen, thanks for the KIP.
> It looks good to me.
> 
> Just a minor suggestion: It would be great if the command asks for a 
> confirmation (y/n) before deleting the topic (similar to how removing ACLs 
> works).
> 

+1 (or some sort of 'force' option to bypass the
query)

[jira] [Commented] (KAFKA-5031) Additional validation in validateMessagesAndAssignOffsets

[ASF GitHub Bot (JIRA)]

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

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

GitHub user gosubpl opened a pull request:

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

KAFKA-5031: validate count of records for DefaultRecordBatch in 
validateMessagesAndAssignOffsets

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

Implements additional check for `DefaultRecordBatch` that compares number 
of records declared in the header with actual number of records.

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

$ git pull https://github.com/gosubpl/kafka KAFKA-5031

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

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

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

This closes #3156


commit 9a14e56846d26471042e0fcab701fecf5f2fccd7
Author: gosubpl 
Date:   2017-05-26T20:12:29Z

KAFKA-5031: validate count of records for DefaultRecordBatch in 
validateMessagesAndAssignOffsets




> Additional validation in validateMessagesAndAssignOffsets
> -
>
> Key: KAFKA-5031
> URL: https://issues.apache.org/jira/browse/KAFKA-5031
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Ismael Juma
>  Labels: exactly-once
> Fix For: 0.11.0.0
>
>
> In validateMessagesAndAssignOffsets(), when validating the 
> DefaultRecordBatch, we should also validate:
> 1. Message count matches the actual number of messages in the array
> 2. The header count matches the actual number of headers



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

[GitHub] kafka pull request #3156: KAFKA-5031: validate count of records for DefaultR...

[gosubpl]

GitHub user gosubpl opened a pull request:

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

KAFKA-5031: validate count of records for DefaultRecordBatch in 
validateMessagesAndAssignOffsets

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

Implements additional check for `DefaultRecordBatch` that compares number 
of records declared in the header with actual number of records.

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

$ git pull https://github.com/gosubpl/kafka KAFKA-5031

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

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

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

This closes #3156


commit 9a14e56846d26471042e0fcab701fecf5f2fccd7
Author: gosubpl 
Date:   2017-05-26T20:12:29Z

KAFKA-5031: validate count of records for DefaultRecordBatch in 
validateMessagesAndAssignOffsets




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

[jira] [Commented] (KAFKA-5329) Replica list in the metadata cache on the broker may have different order from zookeeper

[James Cheng (JIRA)]

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

James Cheng commented on KAFKA-5329:


I noticed this as well. https://github.com/edenhill/kafkacat/issues/98

I was using the results of "kafkacat -L -J" to calculate the preferred leader 
imbalance ratio of my cluster, and my calculation didn't match what the 
controller was doing. I eventually figured out it was because the the replica 
lists weren't in the same order as what was in zookeeper.

> Replica list in the metadata cache on the broker may have different order 
> from zookeeper
> 
>
> Key: KAFKA-5329
> URL: https://issues.apache.org/jira/browse/KAFKA-5329
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.10.2.1
>Reporter: Jiangjie Qin
>  Labels: newbie
> Fix For: 0.11.0.1
>
>
> It looks that in {{PartitionStateInfo}} we are storing the replicas in a set 
> instead of a Seq. This causes the replica order to be lost. In most case it 
> is fine, but in the context of preferred leader election, the replica order 
> determines which replica is the preferred leader of a partition. It would be 
> useful to make the order always be consistent with the order in zookeeper. 



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

[GitHub] kafka-site pull request #56: Update commiter page to indicate that Gwen is a...

[wushujames]

GitHub user wushujames opened a pull request:

https://github.com/apache/kafka-site/pull/56

Update commiter page to indicate that Gwen is a PMC member.

Would it be a conflict of interest for @gwenshap to approve this PR? :)

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

$ git pull https://github.com/wushujames/kafka-site gwen_pmc

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

https://github.com/apache/kafka-site/pull/56.patch

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

This closes #56


commit ddbb5917b1cdc8e233ae5265c8f3276518276cb6
Author: James Cheng 
Date:   2017-05-26T19:40:46Z

Update commiter page to indicate that Gwen is a PMC member.

commit a4bdb1fb0a225c30352ad5bef573a000e3c8482a
Author: James Cheng 
Date:   2017-05-26T19:42:44Z

Spaces instead of tabs.




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

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

[Apache Jenkins Server]

See 

[jira] [Commented] (KAFKA-3151) kafka-consumer-groups.sh fail with sasl enabled

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian commented on KAFKA-3151:


For this to work I had to also add the following line to the properties file 
that [~baluchicken] mentioned:
{code}
sasl.mechanism=PLAIN
{code}

> kafka-consumer-groups.sh fail with sasl enabled 
> 
>
> Key: KAFKA-3151
> URL: https://issues.apache.org/jira/browse/KAFKA-3151
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.1
> Environment: redhat as6.5
>Reporter: linbao111
>
> ./bin/kafka-consumer-groups.sh --new-consumer  --bootstrap-server 
> slave1.otocyon.com:9092 --list
> Error while executing consumer group command Request METADATA failed on 
> brokers List(Node(-1, slave1.otocyon.com, 9092))
> java.lang.RuntimeException: Request METADATA failed on brokers List(Node(-1, 
> slave1.otocyon.com, 9092))
> at kafka.admin.AdminClient.sendAnyNode(AdminClient.scala:73)
> at kafka.admin.AdminClient.findAllBrokers(AdminClient.scala:93)
> at kafka.admin.AdminClient.listAllGroups(AdminClient.scala:101)
> at 
> kafka.admin.AdminClient.listAllGroupsFlattened(AdminClient.scala:122)
> at 
> kafka.admin.AdminClient.listAllConsumerGroupsFlattened(AdminClient.scala:126)
> at 
> kafka.admin.ConsumerGroupCommand$KafkaConsumerGroupService.list(ConsumerGroupCommand.scala:310)
> at 
> kafka.admin.ConsumerGroupCommand$.main(ConsumerGroupCommand.scala:61)
> at kafka.admin.ConsumerGroupCommand.main(ConsumerGroupCommand.scala)
> same error for:
> bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand  --bootstrap-server 
> slave16:9092,app:9092 --describe --group test-consumer-group  --new-consumer
> Error while executing consumer group command Request GROUP_COORDINATOR failed 
> on brokers List(Node(-1, slave16, 9092), Node(-2, app, 9092))
> java.lang.RuntimeException: Request GROUP_COORDINATOR failed on brokers 
> List(Node(-1, slave16, 9092), Node(-2, app, 9092))
> at kafka.admin.AdminClient.sendAnyNode(AdminClient.scala:73)
> at kafka.admin.AdminClient.findCoordinator(AdminClient.scala:78)
> at kafka.admin.AdminClient.describeGroup(AdminClient.scala:130)
> at 
> kafka.admin.AdminClient.describeConsumerGroup(AdminClient.scala:152)
> at 
> kafka.admin.ConsumerGroupCommand$KafkaConsumerGroupService.describeGroup(ConsumerGroupCommand.scala:314)
> at 
> kafka.admin.ConsumerGroupCommand$ConsumerGroupService$class.describe(ConsumerGroupCommand.scala:84)
> at 
> kafka.admin.ConsumerGroupCommand$KafkaConsumerGroupService.describe(ConsumerGroupCommand.scala:302)
> at 
> kafka.admin.ConsumerGroupCommand$.main(ConsumerGroupCommand.scala:63)
> at kafka.admin.ConsumerGroupCommand.main(ConsumerGroupCommand.scala)



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

[jira] [Resolved] (KAFKA-5011) Replica fetchers may need to down-convert messages during a selective message format upgrade

[Jun Rao (JIRA)]

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

Jun Rao resolved KAFKA-5011.

Resolution: Won't Fix

According to Joel's comment, closing this for now.

> Replica fetchers may need to down-convert messages during a selective message 
> format upgrade
> 
>
> Key: KAFKA-5011
> URL: https://issues.apache.org/jira/browse/KAFKA-5011
> Project: Kafka
>  Issue Type: Bug
>Reporter: Joel Koshy
>Assignee: Jiangjie Qin
> Fix For: 0.11.1.0
>
>




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

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

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-5309: Stores not queryable after one thread died

--
[...truncated 904.93 KB...]

kafka.log.BrokerCompressionTest > testBrokerSideCompression[3] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[4] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[4] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[5] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[5] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[6] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[6] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[7] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[7] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[8] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[8] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[9] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[9] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[10] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[10] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[11] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[11] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[12] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[12] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[13] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[13] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[14] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[14] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[15] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[15] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[16] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[16] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[17] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[17] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[18] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[18] PASSED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[19] STARTED

kafka.log.BrokerCompressionTest > testBrokerSideCompression[19] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[0] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[0] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[0] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[0] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[0] STARTED

kafka.log.LogCleanerIntegrationTest > cleanerTest[0] PASSED

kafka.log.LogCleanerIntegrationTest > testCleanerWithMessageFormatV0[0] STARTED

kafka.log.LogCleanerIntegrationTest > testCleanerWithMessageFormatV0[0] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[1] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[1] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[1] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[1] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[1] STARTED

kafka.log.LogCleanerIntegrationTest > cleanerTest[1] PASSED

kafka.log.LogCleanerIntegrationTest > testCleanerWithMessageFormatV0[1] STARTED

kafka.log.LogCleanerIntegrationTest > testCleanerWithMessageFormatV0[1] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[2] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[2] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[2] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[2] PASSED

kafka.log.LogCleanerIntegrationTest > cleanerTest[2] STARTED

kafka.log.LogCleanerIntegrationTest > cleanerTest[2] PASSED

kafka.log.LogCleanerIntegrationTest > testCleanerWithMessageFormatV0[2] STARTED

kafka.log.LogCleanerIntegrationTest > testCleanerWithMessageFormatV0[2] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[3] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleansCombinedCompactAndDeleteTopic[3] PASSED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[3] STARTED

kafka.log.LogCleanerIntegrationTest > 
testCleaningNestedMessagesWithMultipleVersions[3] PASSED

kafka.log.LogCleanerInteg

[jira] [Commented] (KAFKA-5329) Replica list in the metadata cache on the broker may have different order from zookeeper

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-5329:
-

[~ijuma] The ControllerContext is fine. This ticket is not about controller 
context, but the metadata cache in the brokers, which will be returned to the 
user in the TopicMetadataResponse. 

The issue is following. If user wants to assign one of the broker as the leader 
of a partition, they will first have to ensure that broker is the first replica 
in the replica array, then do a preferred leader election for that partition. 
In this case, user can either look into zkNode, or get the replica list from 
the TopicMetadataResponse from brokers. If users are designating a lot of 
leaders, using TopicMetadataResponse would be much more efficient. So it is 
useful to make sure the replica list in the TopicMetadataResponse is correct. 
Also given the metadata returned from broker represents replicas in an array, 
it is weird to not making it align with the actual replica order.

> Replica list in the metadata cache on the broker may have different order 
> from zookeeper
> 
>
> Key: KAFKA-5329
> URL: https://issues.apache.org/jira/browse/KAFKA-5329
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.10.2.1
>Reporter: Jiangjie Qin
>  Labels: newbie
> Fix For: 0.11.0.1
>
>
> It looks that in {{PartitionStateInfo}} we are storing the replicas in a set 
> instead of a Seq. This causes the replica order to be lost. In most case it 
> is fine, but in the context of preferred leader election, the replica order 
> determines which replica is the preferred leader of a partition. It would be 
> useful to make the order always be consistent with the order in zookeeper. 



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

[jira] [Created] (KAFKA-5335) Controller should batch updatePartitionReassignmentData() operation

[Dong Lin (JIRA)]

Dong Lin created KAFKA-5335:
---

 Summary: Controller should batch updatePartitionReassignmentData() 
operation
 Key: KAFKA-5335
 URL: https://issues.apache.org/jira/browse/KAFKA-5335
 Project: Kafka
  Issue Type: Bug
Reporter: Dong Lin
Assignee: Dong Lin


Currently controller will update partition reassignment data every time a 
partition in the reassignment is completed. It means that if user specifies a 
huge reassignment znode of size 1 MB to move 10K partitions, controller will 
need to write roughly 0.5 MB * 1 = 5 GB data to zookeeper in order to 
complete this reassignment. This is because controller needs to write the 
remaining partitions to the znode every time a partition is completely moved.

This is problematic because such a huge reassignment may greatly slow down 
Kafka controller. Note that partition reassignment doesn't necessarily cause 
data movement between brokers because we may use it only to recorder the 
replica list of partitions to evenly distribute preferred leader.



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

Re: KIP-162: Enable topic deletion by default

[Vahid S Hashemian]

Gwen, thanks for the KIP.
It looks good to me.

Just a minor suggestion: It would be great if the command asks for a 
confirmation (y/n) before deleting the topic (similar to how removing ACLs 
works).

Thanks.
--Vahid



From:   Gwen Shapira 
To: "dev@kafka.apache.org" , Users 

Date:   05/26/2017 07:04 AM
Subject:KIP-162: Enable topic deletion by default



Hi Kafka developers, users and friends,

I've added a KIP to improve our out-of-the-box usability a bit:
KIP-162: Enable topic deletion by default:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+Enable+topic+deletion+by+default


Pretty simple :) Discussion and feedback are welcome.

Gwen

[jira] [Commented] (KAFKA-5128) TransactionCoordinator - Check inter broker protocol and message format and raise errors if incompatible

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> TransactionCoordinator - Check inter broker protocol and message format and 
> raise errors if incompatible
> 
>
> Key: KAFKA-5128
> URL: https://issues.apache.org/jira/browse/KAFKA-5128
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Damian Guy
>Assignee: Damian Guy
>Priority: Blocker
>  Labels: exactly-once
> Fix For: 0.11.0.0
>
>
> In {{TransactionStateManager}} and {{InterBrokerSendThread}} we need to check 
> if the message format supports transactions and error out if it doesn't



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

[GitHub] kafka pull request #3103: KAFKA-5128: check inter broker in transactional me...

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Updated] (KAFKA-5128) TransactionCoordinator - Check inter broker protocol and message format and raise errors if incompatible

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-5128:
---
   Resolution: Fixed
Fix Version/s: 0.11.0.0
   Status: Resolved  (was: Patch Available)

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

> TransactionCoordinator - Check inter broker protocol and message format and 
> raise errors if incompatible
> 
>
> Key: KAFKA-5128
> URL: https://issues.apache.org/jira/browse/KAFKA-5128
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Damian Guy
>Assignee: Damian Guy
>Priority: Blocker
>  Labels: exactly-once
> Fix For: 0.11.0.0
>
>
> In {{TransactionStateManager}} and {{InterBrokerSendThread}} we need to check 
> if the message format supports transactions and error out if it doesn't



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

[jira] [Commented] (KAFKA-5309) Stores not queryable after one thread died

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Stores not queryable after one thread died
> --
>
> Key: KAFKA-5309
> URL: https://issues.apache.org/jira/browse/KAFKA-5309
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.1
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
> Fix For: 0.11.0.0
>
>
> For a multi threaded Streams instance, it can happen that one thread dies 
> while all other threads (within a single instance) keep running. Even if this 
> is an rare scenario, we should allow querying the stores after rebalance 
> finished. However, this does never happen, as the died thread's state is 
> still in {{KafkaStreams}} thread state map (as {{NOT_RUNNING}}), and thus, 
> {{KafkaStreams}} itself is in state {{REBALANCING}} all the time and does not 
> transit back to {{RUNNING}}.



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

[GitHub] kafka pull request #3140: KAFKA-5309: Stores not queryable after one thread ...

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Updated] (KAFKA-5309) Stores not queryable after one thread died

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-5309:
-
Resolution: Fixed
Status: Resolved  (was: Patch Available)

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

> Stores not queryable after one thread died
> --
>
> Key: KAFKA-5309
> URL: https://issues.apache.org/jira/browse/KAFKA-5309
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.1
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
> Fix For: 0.11.0.0
>
>
> For a multi threaded Streams instance, it can happen that one thread dies 
> while all other threads (within a single instance) keep running. Even if this 
> is an rare scenario, we should allow querying the stores after rebalance 
> finished. However, this does never happen, as the died thread's state is 
> still in {{KafkaStreams}} thread state map (as {{NOT_RUNNING}}), and thus, 
> {{KafkaStreams}} itself is in state {{REBALANCING}} all the time and does not 
> transit back to {{RUNNING}}.



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

[jira] [Commented] (KAFKA-5322) Resolve AddPartitions response error code inconsistency

[Guozhang Wang (JIRA)]

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

Guozhang Wang commented on KAFKA-5322:
--

[~hachikuji] I'd prefer option 2, to still have a per-partition error code as 
it would let the client to handle programmably; otherwise as you mention client 
need to parse the error message string which is more awkward. I think it is 
Okay add another error code like NOT_ATTEMPTED which also leaves the door open 
to partially handle the request on the broker side in the future.

> Resolve AddPartitions response error code inconsistency
> ---
>
> Key: KAFKA-5322
> URL: https://issues.apache.org/jira/browse/KAFKA-5322
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>Priority: Blocker
> Fix For: 0.11.0.0
>
>
> The AddPartitions request does not support partial failures. Either all 
> partitions are successfully added to the transaction or none of them are. 
> Currently we return a separate error code for each partition that was added 
> to the transaction, which begs the question of what error code to return if 
> we have not actually encountered a fatal partition-level error for some 
> partition. For example, suppose we send AddPartitions with partitions A and 
> B. If A is not authorized, we will not even attempt to add B to the 
> transaction, but what error code should we use. The current solution is to 
> only include partition A and its error code in the response, but this is a 
> little inconsistent with most other request types. Alternatives that have 
> been proposed:
> 1. Instead of a partition-level error, use one global error. We can add a 
> global error message to return friendlier details to the user about which 
> partition had a fault. The downside is that we would have to parse the 
> message contents if we wanted to do any partition-specific handling. We could 
> not easily fill the set of topics in {{TopicAuthorizationException}} for 
> example.
> 2. We can add a new error code to indicate that the broker did not even 
> attempt to add the partition to the transaction. For example: 
> OPERATION_NOT_ATTEMPTED or something like that. 



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

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Damian Guy]

In that case, though, every access to that key is doomed to failure as the
database is corrupted. So i think it should probably die in a steaming heap
at that point!

On Fri, 26 May 2017 at 17:33 Eno Thereska  wrote:

> Hi Damian,
>
> I was thinking of cases when there is bit-rot on the storage itself and we
> get a malformed record that cannot be de-serialized. There is an
> interesting intersection here with CRCs in both Kafka (already there, they
> throw on deserialization) and potentially local storage (we don't have CRCs
> here on the data files, though RocksDB has them on its write-ahead log
> records).
>
> Basically in a nutshell, I'm saying that every deserialization exception
> should go through this new path. The user can decide to fail or continue.
> We could start with just poison pills from Kafka though and punt the
> storage one to later.
>
> Eno
>
> > On 26 May 2017, at 16:59, Damian Guy  wrote:
> >
> > Eno,
> >
> > Under what circumstances would you get a deserialization exception from
> the
> > state store? I can only think of the case where someone has provided a
> bad
> > deserializer to a method that creates a state store. In which case it
> would
> > be a user error and probably should just abort?
> >
> > Thanks,
> > Damian
> >
> > On Fri, 26 May 2017 at 16:32 Eno Thereska 
> wrote:
> >
> >> See latest reply to Jan's note. I think I unnecessarily broadened the
> >> scope of this KIP to the point where it sounded like it handles all
> sorts
> >> of exceptions. The scope should be strictly limited to "poison pill"
> >> records for now. Will update KIP,
> >>
> >> Thanks
> >> Eno
> >>> On 26 May 2017, at 16:16, Matthias J. Sax 
> wrote:
> >>>
> >>> "bad" for this case would mean, that we got an
> >>> `DeserializationException`. I am not sure if any other processing error
> >>> should be covered?
> >>>
> >>> @Eno: this raises one one question. Might it be better to allow for two
> >>> handlers instead of one? One for deserialization exception and one for
> >>> all other exceptions from user code?
> >>>
> >>> Just a thought.
> >>>
> >>>
> >>> -Matthias
> >>>
> >>> On 5/26/17 7:49 AM, Jim Jagielski wrote:
> 
> > On May 26, 2017, at 5:13 AM, Eno Thereska 
> >> wrote:
> >
> >
> >>
> >>
> >> With regard to `DeserializationException`, do you thing it might
> make
> >> sense to have a "dead letter queue" as a feature to provide
> >> out-of-the-box?
> >
> > We could provide a special topic where bad messages go to, and then
> >> we'd have to add a config option for the user to provide a topic. Is
> that
> >> what you're thinking?
> >
> 
>  For various definitions of "bad"??
> 
> >>>
> >>
> >>
>
>

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Eno Thereska]

Hi Damian,

I was thinking of cases when there is bit-rot on the storage itself and we get 
a malformed record that cannot be de-serialized. There is an interesting 
intersection here with CRCs in both Kafka (already there, they throw on 
deserialization) and potentially local storage (we don't have CRCs here on the 
data files, though RocksDB has them on its write-ahead log records). 

Basically in a nutshell, I'm saying that every deserialization exception should 
go through this new path. The user can decide to fail or continue. We could 
start with just poison pills from Kafka though and punt the storage one to 
later. 

Eno

> On 26 May 2017, at 16:59, Damian Guy  wrote:
> 
> Eno,
> 
> Under what circumstances would you get a deserialization exception from the
> state store? I can only think of the case where someone has provided a bad
> deserializer to a method that creates a state store. In which case it would
> be a user error and probably should just abort?
> 
> Thanks,
> Damian
> 
> On Fri, 26 May 2017 at 16:32 Eno Thereska  wrote:
> 
>> See latest reply to Jan's note. I think I unnecessarily broadened the
>> scope of this KIP to the point where it sounded like it handles all sorts
>> of exceptions. The scope should be strictly limited to "poison pill"
>> records for now. Will update KIP,
>> 
>> Thanks
>> Eno
>>> On 26 May 2017, at 16:16, Matthias J. Sax  wrote:
>>> 
>>> "bad" for this case would mean, that we got an
>>> `DeserializationException`. I am not sure if any other processing error
>>> should be covered?
>>> 
>>> @Eno: this raises one one question. Might it be better to allow for two
>>> handlers instead of one? One for deserialization exception and one for
>>> all other exceptions from user code?
>>> 
>>> Just a thought.
>>> 
>>> 
>>> -Matthias
>>> 
>>> On 5/26/17 7:49 AM, Jim Jagielski wrote:
 
> On May 26, 2017, at 5:13 AM, Eno Thereska 
>> wrote:
> 
> 
>> 
>> 
>> With regard to `DeserializationException`, do you thing it might make
>> sense to have a "dead letter queue" as a feature to provide
>> out-of-the-box?
> 
> We could provide a special topic where bad messages go to, and then
>> we'd have to add a config option for the user to provide a topic. Is that
>> what you're thinking?
> 
 
 For various definitions of "bad"??
 
>>> 
>> 
>> 

Re: Kafka Loss of messages

[Jun Rao]

Hi, Vinayak,

I am not sure that KAFKA-1561 as reported is an issue. Could you describe
the scenario in your test in that jira in a bit more detail?

Thanks,

Jun

On Fri, May 26, 2017 at 5:58 AM, Vinayak Sharma  wrote:

> Hi,
>
> I am experiencing loss of messages while scaling brokers up and down in
> kafka cluster.
>
> I came across this jira issue(link
> ). Can you confirm if
> this is actually an issue in kafka or can the problem be solved by changing
> some configuration parameters.
>
> I am enclosing the server.properties that I am using.
>
> Please get to back to me.
>
> Regards,
> Vinayak.
>
> 

[jira] [Commented] (KAFKA-1561) Data Loss for Incremented Replica Factor and Leader Election

[Jun Rao (JIRA)]

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

Jun Rao commented on KAFKA-1561:


Hmm, interesting. From the description, when the LEOs on the leader are 574 and 
571, the HW on broker 0 is still at 571. This suggests that messages between 
572 and 574 haven't been committed and the producer shouldn't have received a 
successful ack with acks=-1.

> Data Loss for Incremented Replica Factor and Leader Election
> 
>
> Key: KAFKA-1561
> URL: https://issues.apache.org/jira/browse/KAFKA-1561
> Project: Kafka
>  Issue Type: Bug
>Reporter: Guozhang Wang
>Assignee: Guozhang Wang
> Attachments: broker0.log, broker2.log, consumer.log, producer.log
>
>
> This is reported on the mailing list (thanks to Jad).
> {quote}
> Hi,
> I have a test that continuously sends messages to one broker, brings up
> another broker, and adds it as a replica for all partitions, with it being
> the preferred replica for some. I have auto.leader.rebalance.enable=true,
> so replica election gets triggered. Data is being pumped to the old broker
> all the while. It seems that some data gets lost while switching over to
> the new leader. Is this a bug, or do I have something misconfigured? I also
> have request.required.acks=-1 on the producer.
> Here's what I think is happening:
> 1. Producer writes message to broker 0, [EventServiceUpsertTopic,13], w/
> broker 0 currently leader, with ISR=(0), so write returns successfully,
> even when acks = -1. Correlation id 35836
> Producer log:
> [2014-07-24 14:44:26,991]  [DEBUG]  [dw-97 - PATCH
> /v1/events/type_for_test_bringupNewBroker_shouldRebalance_shouldNotLoseData/event?_idPath=idField&_mergeFields=field1]
> [kafka.producer.BrokerPartitionInfo]  Partition
> [EventServiceUpsertTopic,13] has leader 0
> [2014-07-24 14:44:26,993]  [DEBUG]  [dw-97 - PATCH
> /v1/events/type_for_test_bringupNewBroker_shouldRebalance_shouldNotLoseData/event?_idPath=idField&_mergeFields=field1]
> [k.producer.async.DefaultEventHandler]  Producer sent messages with
> correlation id 35836 for topics [EventServiceUpsertTopic,13] to broker 0 on
> localhost:56821
> 2. Broker 1 is still catching up
> Broker 0 Log:
> [2014-07-24 14:44:26,992]  [DEBUG]  [kafka-request-handler-3]
> [kafka.cluster.Partition]  Partition [EventServiceUpsertTopic,13] on broker
> 0: Old hw for partition [EventServiceUpsertTopic,13] is 971. New hw is 971.
> All leo's are 975,971
> [2014-07-24 14:44:26,992]  [DEBUG]  [kafka-request-handler-3]
> [kafka.server.KafkaApis]  [KafkaApi-0] Produce to local log in 0 ms
> [2014-07-24 14:44:26,992]  [DEBUG]  [kafka-processor-56821-0]
> [kafka.request.logger]  Completed request:Name: ProducerRequest; Version:
> 0; CorrelationId: 35836; ClientId: ; RequiredAcks: -1; AckTimeoutMs: 1
> ms from client /127.0.0.1:57086
> ;totalTime:0,requestQueueTime:0,localTime:0,remoteTime:0,responseQueueTime:0,sendTime:0
> 3. Leader election is triggered by the scheduler:
> Broker 0 Log:
> [2014-07-24 14:44:26,991]  [INFO ]  [kafka-scheduler-0]
> [k.c.PreferredReplicaPartitionLeaderSelector]
> [PreferredReplicaPartitionLeaderSelector]: Current leader 0 for partition [
> EventServiceUpsertTopic,13] is not the preferred replica. Trigerring
> preferred replica leader election
> [2014-07-24 14:44:26,993]  [DEBUG]  [kafka-scheduler-0]
> [kafka.utils.ZkUtils$]  Conditional update of path
> /brokers/topics/EventServiceUpsertTopic/partitions/13/state with value
> {"controller_epoch":1,"leader":1,"version":1,"leader_epoch":3,"isr":[0,1]}
> and expected version 3 succeeded, returning the new version: 4
> [2014-07-24 14:44:26,994]  [DEBUG]  [kafka-scheduler-0]
> [k.controller.PartitionStateMachine]  [Partition state machine on
> Controller 0]: After leader election, leader cache is updated to
> Map((Leader:1,ISR:0,1,LeaderEpoch:3,ControllerEpoch:1),)
> [2014-07-24 14:44:26,994]  [INFO ]  [kafka-scheduler-0]
> [kafka.controller.KafkaController]  [Controller 0]: Partition [
> EventServiceUpsertTopic,13] completed preferred replica leader election.
> New leader is 1
> 4. Broker 1 is still behind, but it sets the high water mark to 971!!!
> Broker 1 Log:
> [2014-07-24 14:44:26,999]  [INFO ]  [kafka-request-handler-6]
> [kafka.server.ReplicaFetcherManager]  [ReplicaFetcherManager on broker 1]
> Removed fetcher for partitions [EventServiceUpsertTopic,13]
> [2014-07-24 14:44:27,000]  [DEBUG]  [kafka-request-handler-6]
> [kafka.cluster.Partition]  Partition [EventServiceUpsertTopic,13] on broker
> 1: Old hw for partition [EventServiceUpsertTopic,13] is 970. New hw is -1.
> All leo's are -1,971
> [2014-07-24 14:44:27,098]  [DEBUG]  [kafka-request-handler-3]
> [kafka.server.KafkaApis]  [KafkaApi-1] Maybe update partition HW due to
> fetch request: Na

Kafka Loss of messages

[Vinayak Sharma]

Hi,

I am experiencing loss of messages while scaling brokers up and down in
kafka cluster.

I came across this jira issue(link
). Can you confirm if
this is actually an issue in kafka or can the problem be solved by changing
some configuration parameters.

I am enclosing the server.properties that I am using.

Please get to back to me.

Regards,
Vinayak.

-- 
 

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Damian Guy]

Eno,

Under what circumstances would you get a deserialization exception from the
state store? I can only think of the case where someone has provided a bad
deserializer to a method that creates a state store. In which case it would
be a user error and probably should just abort?

Thanks,
Damian

On Fri, 26 May 2017 at 16:32 Eno Thereska  wrote:

> See latest reply to Jan's note. I think I unnecessarily broadened the
> scope of this KIP to the point where it sounded like it handles all sorts
> of exceptions. The scope should be strictly limited to "poison pill"
> records for now. Will update KIP,
>
> Thanks
> Eno
> > On 26 May 2017, at 16:16, Matthias J. Sax  wrote:
> >
> > "bad" for this case would mean, that we got an
> > `DeserializationException`. I am not sure if any other processing error
> > should be covered?
> >
> > @Eno: this raises one one question. Might it be better to allow for two
> > handlers instead of one? One for deserialization exception and one for
> > all other exceptions from user code?
> >
> > Just a thought.
> >
> >
> > -Matthias
> >
> > On 5/26/17 7:49 AM, Jim Jagielski wrote:
> >>
> >>> On May 26, 2017, at 5:13 AM, Eno Thereska 
> wrote:
> >>>
> >>>
> 
> 
>  With regard to `DeserializationException`, do you thing it might make
>  sense to have a "dead letter queue" as a feature to provide
> out-of-the-box?
> >>>
> >>> We could provide a special topic where bad messages go to, and then
> we'd have to add a config option for the user to provide a topic. Is that
> what you're thinking?
> >>>
> >>
> >> For various definitions of "bad"??
> >>
> >
>
>

[jira] [Updated] (KAFKA-3514) Stream timestamp computation needs some further thoughts

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-3514:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Stream timestamp computation needs some further thoughts
> 
>
> Key: KAFKA-3514
> URL: https://issues.apache.org/jira/browse/KAFKA-3514
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Eno Thereska
>  Labels: architecture
> Fix For: 0.11.1.0
>
>
> Our current stream task's timestamp is used for punctuate function as well as 
> selecting which stream to process next (i.e. best effort stream 
> synchronization). And it is defined as the smallest timestamp over all 
> partitions in the task's partition group. This results in two unintuitive 
> corner cases:
> 1) observing a late arrived record would keep that stream's timestamp low for 
> a period of time, and hence keep being process until that late record. For 
> example take two partitions within the same task annotated by their 
> timestamps:
> {code}
> Stream A: 5, 6, 7, 8, 9, 1, 10
> {code}
> {code}
> Stream B: 2, 3, 4, 5
> {code}
> The late arrived record with timestamp "1" will cause stream A to be selected 
> continuously in the thread loop, i.e. messages with timestamp 5, 6, 7, 8, 9 
> until the record itself is dequeued and processed, then stream B will be 
> selected starting with timestamp 2.
> 2) an empty buffered partition will cause its timestamp to be not advanced, 
> and hence the task timestamp as well since it is the smallest among all 
> partitions. This may not be a severe problem compared with 1) above though.
> *Update*
> There is one more thing to consider (full discussion found here: 
> http://search-hadoop.com/m/Kafka/uyzND1iKZJN1yz0E5?subj=Order+of+punctuate+and+process+in+a+stream+processor)
> {quote}
> Let's assume the following case.
> - a stream processor that uses the Processor API
> - context.schedule(1000) is called in the init()
> - the processor reads only one topic that has one partition
> - using custom timestamp extractor, but that timestamp is just a wall 
> clock time
> Image the following events:
> 1., for 10 seconds I send in 5 messages / second
> 2., does not send any messages for 3 seconds
> 3., starts the 5 messages / second again
> I see that punctuate() is not called during the 3 seconds when I do not 
> send any messages. This is ok according to the documentation, because 
> there is not any new messages to trigger the punctuate() call. When the 
> first few messages arrives after a restart the sending (point 3. above) I 
> see the following sequence of method calls:
> 1., process() on the 1st message
> 2., punctuate() is called 3 times
> 3., process() on the 2nd message
> 4., process() on each following message
> What I would expect instead is that punctuate() is called first and then 
> process() is called on the messages, because the first message's timestamp 
> is already 3 seconds older then the last punctuate() was called, so the 
> first message belongs after the 3 punctuate() calls.
> {quote}



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

[jira] [Updated] (KAFKA-4641) Improve test coverage of StreamsThread

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4641:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage of StreamsThread
> --
>
> Key: KAFKA-4641
> URL: https://issues.apache.org/jira/browse/KAFKA-4641
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Damian Guy
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> Some methods in {{StreamThread}} have little or no coverage.
> In particular:
> {{maybeUpdateStandbyTasks}} has little to no coverage
> Committing of StandbyTasks in {{commitAll}}
> {{maybePunctuate}}
> {{commitOne}} - no tests for exceptions
> {{unAssignChangeLogPartitions} - no tests for exceptions
> {{addStreamsTask}} - no tests for exceptions
> {{runLoop}}
> Please see coverage report attached to parent



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

[jira] [Updated] (KAFKA-3779) Add the LRU cache for KTable.to() operator

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-3779:

Fix Version/s: (was: 0.11.0.0)

> Add the LRU cache for KTable.to() operator
> --
>
> Key: KAFKA-3779
> URL: https://issues.apache.org/jira/browse/KAFKA-3779
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.1.0
>Reporter: Eno Thereska
>
> The KTable.to operator currently does not use a cache. We can add a cache to 
> this operator to deduplicate and reduce data traffic as well. This is to be 
> done after KAFKA-3777.



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

[jira] [Updated] (KAFKA-4640) Improve Streams unit test coverage

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4640:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve Streams unit test coverage
> --
>
> Key: KAFKA-4640
> URL: https://issues.apache.org/jira/browse/KAFKA-4640
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Damian Guy
>Assignee: Damian Guy
>Priority: Minor
> Fix For: 0.11.1.0
>
> Attachments: streams-coverage.zip
>
>
> There are some important methods in streams that are lacking good unit-test 
> coverage. Whilst we shouldn't strive to get 100% coverage, we should do our 
> best to ensure sure that all important code paths are covered by unit-tests.



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

[jira] [Updated] (KAFKA-4653) Improve test coverage of RocksDBStore

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4653:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage of RocksDBStore
> -
>
> Key: KAFKA-4653
> URL: https://issues.apache.org/jira/browse/KAFKA-4653
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> {{putAll}} - not covered
> {{putInternal} - exceptions



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

[jira] [Updated] (KAFKA-4643) Improve test coverage of StreamsKafkaClient

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4643:

Fix Version/s: (was: 0.11.0.0)

> Improve test coverage of StreamsKafkaClient
> ---
>
> Key: KAFKA-4643
> URL: https://issues.apache.org/jira/browse/KAFKA-4643
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>
> Exception paths not tested.
> {{getTopicMetadata}} not tested



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

[jira] [Updated] (KAFKA-4651) Improve test coverage of Stores

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4651:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage of Stores
> ---
>
> Key: KAFKA-4651
> URL: https://issues.apache.org/jira/browse/KAFKA-4651
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> Some factory methods aren't tested



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

[jira] [Updated] (KAFKA-4650) Improve test coverage org.apache.kafka.streams.kstream.internals

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4650:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage org.apache.kafka.streams.kstream.internals
> 
>
> Key: KAFKA-4650
> URL: https://issues.apache.org/jira/browse/KAFKA-4650
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Damian Guy
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> Lots of classes have little or no coverage at all, i.e., 
> {{KTableAggregate.KTableAggregateValueGetter}}
> {{KTableKTableRightJoin.KTableKTableRightJoinValueGetterSupplier}}
> {{KStreamAggregate.KStreamAggregateValueGetter}}
> {{KStreamReduce.KStreamReduceValueGetter}}
> {{KStreamWindowReduce.new KTableValueGetterSupplier}}
> {{KTableAggregate.new KTableValueGetterSupplier}}
> {{KTableRepartitionMap.new KTableValueGetterSupplier}}
> {{KTableKTableRightJoin.KTableKTableRightJoinValueGetter}}
> {{KTableKTableLeftJoinValueGetter}}
> {{KStreamWindowReduce.KStreamWindowReduceValueGetter}}
> {{TimeWindow}}
> {{ChangedSerializer}}
> {{UnlimitedWindow}}
> {{WindowedDeserializer}}
> {{KStreamSessionWindowAggregate.KTableSessionWindowValueGetter}}
> {{KTableRepartitionMap}}



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

[jira] [Updated] (KAFKA-4655) Improve test coverage of CompositeReadOnlySessionStore

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4655:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage of CompositeReadOnlySessionStore
> --
>
> Key: KAFKA-4655
> URL: https://issues.apache.org/jira/browse/KAFKA-4655
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
> Fix For: 0.11.1.0
>
>
> exceptions in fetch and internal iterator



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

[jira] [Updated] (KAFKA-4658) Improve test coverage InMemoryKeyValueLoggedStore

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4658:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage InMemoryKeyValueLoggedStore
> -
>
> Key: KAFKA-4658
> URL: https://issues.apache.org/jira/browse/KAFKA-4658
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
> Fix For: 0.11.1.0
>
>
> {{putAll} not covered



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

[jira] [Updated] (KAFKA-4659) Improve test coverage of CachingKeyValueStore

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4659:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage of CachingKeyValueStore
> -
>
> Key: KAFKA-4659
> URL: https://issues.apache.org/jira/browse/KAFKA-4659
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> {{putIfAbsent}} mostly not covered



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

[jira] [Updated] (KAFKA-4656) Improve test coverage of CompositeReadOnlyKeyValueStore

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4656:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve test coverage of CompositeReadOnlyKeyValueStore
> ---
>
> Key: KAFKA-4656
> URL: https://issues.apache.org/jira/browse/KAFKA-4656
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> exceptions not covered



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

[jira] [Updated] (KAFKA-4730) Streams does not have an in-memory windowed store

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4730:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Streams does not have an in-memory windowed store
> -
>
> Key: KAFKA-4730
> URL: https://issues.apache.org/jira/browse/KAFKA-4730
> Project: Kafka
>  Issue Type: New Feature
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
>Assignee: Nikki Thean
> Fix For: 0.11.1.0
>
>
> Streams has windowed persistent stores (e.g., see PersistentKeyValueFactory 
> interface with "windowed" method), however it does not allow for windowed 
> in-memory stores (e.g., see InMemoryKeyValueFactory interface). 
> In addition to the interface not allowing it, streams does not actually have 
> an implementation of an in-memory windowed store.
> The implications are that operations that require windowing cannot use 
> in-memory stores. 



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

[jira] [Updated] (KAFKA-5037) Infinite loop if all input topics are unknown at startup

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax updated KAFKA-5037:
---
Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Infinite loop if all input topics are unknown at startup
> 
>
> Key: KAFKA-5037
> URL: https://issues.apache.org/jira/browse/KAFKA-5037
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>  Labels: user-experience
> Fix For: 0.11.1.0
>
>
> See discusion: https://github.com/apache/kafka/pull/2815
> We will need some rewrite on {{StreamPartitionsAssignor}} and to add much 
> more test for all kind of corner cases, including pattern subscriptions.



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

[jira] [Updated] (KAFKA-4661) Improve test coverage UsePreviousTimeOnInvalidTimestamp

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4661:

Fix Version/s: (was: 0.11.0.0)

> Improve test coverage UsePreviousTimeOnInvalidTimestamp
> ---
>
> Key: KAFKA-4661
> URL: https://issues.apache.org/jira/browse/KAFKA-4661
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Priority: Minor
>
> Exception branch not tested



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

[jira] [Updated] (KAFKA-4696) Streams standby task assignment should be state-store aware

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4696:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Streams standby task assignment should be state-store aware
> ---
>
> Key: KAFKA-4696
> URL: https://issues.apache.org/jira/browse/KAFKA-4696
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0, 0.11.0.0
>Reporter: Damian Guy
> Fix For: 0.11.1.0
>
>
> Task Assignment is currently not aware of which tasks have State Stores. This 
> can result in uneven balance of standby task assignment as all tasks are 
> assigned, but only those tasks with state-stores are ever created by 
> {{StreamThread}}. So what seems like an optimal strategy during assignment 
> time could be sub-optimal post-assignment.
> For example, lets say we have 4 tasks (2 with state-stores), 2 clients, 
> numStandbyReplicas = 1. Each client would get 2 active and 2 standby tasks.  
> One of the clients may end up with both state-store tasks, while the other 
> has none.
> Further to this, standby task configuration is currently "all or nothing". It 
> might make sense to allow more fine grained configuration, i.e., the ability 
> to specify the number of standby replicas individually for each stateful 
> operator.



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

[jira] [Updated] (KAFKA-4729) Stores for kstream-kstream join cannot be in-memory

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4729:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Stores for kstream-kstream join cannot be in-memory
> ---
>
> Key: KAFKA-4729
> URL: https://issues.apache.org/jira/browse/KAFKA-4729
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
> Fix For: 0.11.1.0
>
>
> Whereas we can specify in the DSL that stores used for aggregates can be 
> RocksDb-based on in-memory, we cannot do that for stores used for 
> KStream-KStream joins. E.g., the join() methon in KStreamImpl.java creates 
> two state stores and the user does not have the option of having them be 
> in-memory:
> StateStoreSupplier thisWindow =
> createWindowedStateStore(windows, keySerde, lhsValueSerde, 
> joinThisName + "-store");
> StateStoreSupplier otherWindow =
> createWindowedStateStore(windows, keySerde, otherValueSerde, 
> joinOtherName + "-store");
> Part of the problem is that for joins, stores are not exposed to the user. We 
> might want to rethink that.



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

[jira] [Updated] (KAFKA-5063) Flaky ResetIntegrationTest.testReprocessingFromScratchAfterResetWithIntermediateUserTopic

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax updated KAFKA-5063:
---
Fix Version/s: (was: 0.11.1.0)

> Flaky 
> ResetIntegrationTest.testReprocessingFromScratchAfterResetWithIntermediateUserTopic
> -
>
> Key: KAFKA-5063
> URL: https://issues.apache.org/jira/browse/KAFKA-5063
> Project: Kafka
>  Issue Type: Bug
>  Components: streams, unit tests
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>  Labels: unit-test
>
> {noformat}
> org.apache.kafka.streams.integration.ResetIntegrationTest > 
> testReprocessingFromScratchAfterResetWithIntermediateUserTopic FAILED
> java.lang.AssertionError: 
> Expected: <[KeyValue(2983939126775, 1), KeyValue(2983939126815, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 1), KeyValue(2983939126855, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 2), KeyValue(2983939126835, 2), 
> KeyValue(2983939126855, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126835, 2), KeyValue(2983939126855, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126855, 2), KeyValue(2983939126875, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126915, 1), 
> KeyValue(2983939126875, 2), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 1), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 3), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 2), 
> KeyValue(2983939126875, 4), KeyValue(2983939126895, 3), 
> KeyValue(2983939126915, 3), KeyValue(2983939126935, 2), 
> KeyValue(2983939126875, 4), KeyValue(2983939126895, 4), 
> KeyValue(2983939126915, 3), KeyValue(2983939126935, 3)]>
>  but: was <[KeyValue(2983939126775, 1), KeyValue(2983939126815, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 1), KeyValue(2983939126855, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 2), KeyValue(2983939126835, 2), 
> KeyValue(2983939126855, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126835, 2), KeyValue(2983939126855, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126855, 2), KeyValue(2983939126875, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126915, 1), 
> KeyValue(2983939126875, 2), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 1), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 3), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 2), 
> KeyValue(2983939126875, 4)]>
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:8)
> at 
> org.apache.kafka.streams.integration.ResetIntegrationTest.testReprocessingFromScratchAfterResetWithIntermediateUserTopic(ResetIntegrationTest.java:190)
> {noformat}



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

[jira] [Updated] (KAFKA-4875) Kafka Streams: topic groups and builder.stream API

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4875:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Kafka Streams: topic groups and builder.stream API
> --
>
> Key: KAFKA-4875
> URL: https://issues.apache.org/jira/browse/KAFKA-4875
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
> Fix For: 0.11.1.0
>
>
> One thing that has come up in mailing list is that the notion of topic groups 
> is unclear. So if you have 2 topics, each with 3 partitions, you'd expect 6 
> tasks. However, if you do `builder.stream(topic1, topic2)` you actually get 
> only 3 tasks created. If you do `builder.stream(topic1); 
> builder.stream(topic2)` you get 6 tasks, i.e., parallelism is increased. So 
> the same application, calling builder.stream() in two different ways, might 
> see different performance.
> In the Kafka Streams documentations we mention partitions and tasks, but not 
> topic groups. We also do not document the effects of using builder.stream 
> with a topic array. We also need to revisit whether the API and its effects 
> are confusing.



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

[jira] [Resolved] (KAFKA-4838) Export buffer cache metrics in JMX

[Eno Thereska (JIRA)]

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

Eno Thereska resolved KAFKA-4838.
-
Resolution: Won't Fix

> Export buffer cache metrics in JMX
> --
>
> Key: KAFKA-4838
> URL: https://issues.apache.org/jira/browse/KAFKA-4838
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
>Priority: Minor
>  Labels: needs-kip
> Fix For: 0.11.0.0
>
>
> Currently we log certain cache metrics, but don't export those via JMX, e.g., 
> in ThreadCache.flush() we have:
> log.debug("Thread {} cache stats on flush: #puts={}, #gets={}, #evicts={}, 
> #flushes={}",name, puts(), gets(), evicts(), flushes());
> We should export the hits, misses etc as metrics (with a DEBUG recording 
> level). 



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

[jira] [Updated] (KAFKA-3856) Cleanup Kafka Streams builder API

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax updated KAFKA-3856:
---
Fix Version/s: 0.11.1.0

> Cleanup Kafka Streams builder API
> -
>
> Key: KAFKA-3856
> URL: https://issues.apache.org/jira/browse/KAFKA-3856
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Matthias J. Sax
>  Labels: api, kip
> Fix For: 0.11.1.0
>
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-120%3A+Cleanup+Kafka+Streams+builder+API



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

[jira] [Updated] (KAFKA-4904) Performance of RocksDb with state record cache

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4904:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Performance of RocksDb with state record cache
> --
>
> Key: KAFKA-4904
> URL: https://issues.apache.org/jira/browse/KAFKA-4904
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
> Fix For: 0.11.1.0
>
>
> The performance of RocksDb without any record cache is slightly better than 
> with the default of 10MB of the default cache. This needs investigating. It 
> is likely to be the case that this is not an entirely apples-to-apples 
> comparison since the record cache holds records for a maximum of 
> 'commit.time.ms', which by default is 30 seconds. So the record cache is 
> adding quite a bit of latency, and we know that, however documenting the 
> tradeoff and looking if there is any other bugs needs to be done.



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

[jira] [Updated] (KAFKA-4934) Add streams test with RocksDb failures

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-4934:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Add streams test with RocksDb failures
> --
>
> Key: KAFKA-4934
> URL: https://issues.apache.org/jira/browse/KAFKA-4934
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
>Priority: Critical
> Fix For: 0.11.1.0
>
>
> We need to add either integration of system tests with RocksDb failing 
> underneath and fix any problems that occur, including deadlocks, wrong 
> exceptions, etc.



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

[jira] [Updated] (KAFKA-5063) Flaky ResetIntegrationTest.testReprocessingFromScratchAfterResetWithIntermediateUserTopic

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax updated KAFKA-5063:
---
Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Flaky 
> ResetIntegrationTest.testReprocessingFromScratchAfterResetWithIntermediateUserTopic
> -
>
> Key: KAFKA-5063
> URL: https://issues.apache.org/jira/browse/KAFKA-5063
> Project: Kafka
>  Issue Type: Bug
>  Components: streams, unit tests
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>  Labels: unit-test
>
> {noformat}
> org.apache.kafka.streams.integration.ResetIntegrationTest > 
> testReprocessingFromScratchAfterResetWithIntermediateUserTopic FAILED
> java.lang.AssertionError: 
> Expected: <[KeyValue(2983939126775, 1), KeyValue(2983939126815, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 1), KeyValue(2983939126855, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 2), KeyValue(2983939126835, 2), 
> KeyValue(2983939126855, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126835, 2), KeyValue(2983939126855, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126855, 2), KeyValue(2983939126875, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126915, 1), 
> KeyValue(2983939126875, 2), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 1), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 3), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 2), 
> KeyValue(2983939126875, 4), KeyValue(2983939126895, 3), 
> KeyValue(2983939126915, 3), KeyValue(2983939126935, 2), 
> KeyValue(2983939126875, 4), KeyValue(2983939126895, 4), 
> KeyValue(2983939126915, 3), KeyValue(2983939126935, 3)]>
>  but: was <[KeyValue(2983939126775, 1), KeyValue(2983939126815, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 1), KeyValue(2983939126855, 1), 
> KeyValue(2983939126835, 1), KeyValue(2983939126795, 1), 
> KeyValue(2983939126815, 2), KeyValue(2983939126835, 2), 
> KeyValue(2983939126855, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126835, 2), KeyValue(2983939126855, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126875, 1), 
> KeyValue(2983939126855, 2), KeyValue(2983939126875, 2), 
> KeyValue(2983939126895, 1), KeyValue(2983939126915, 1), 
> KeyValue(2983939126875, 2), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 1), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 2), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 1), 
> KeyValue(2983939126875, 3), KeyValue(2983939126895, 3), 
> KeyValue(2983939126915, 2), KeyValue(2983939126935, 2), 
> KeyValue(2983939126875, 4)]>
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:8)
> at 
> org.apache.kafka.streams.integration.ResetIntegrationTest.testReprocessingFromScratchAfterResetWithIntermediateUserTopic(ResetIntegrationTest.java:190)
> {noformat}



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

[jira] [Commented] (KAFKA-5037) Infinite loop if all input topics are unknown at startup

[Eno Thereska (JIRA)]

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

Eno Thereska commented on KAFKA-5037:
-

[~mjsax] is this still for 0.11?

> Infinite loop if all input topics are unknown at startup
> 
>
> Key: KAFKA-5037
> URL: https://issues.apache.org/jira/browse/KAFKA-5037
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>  Labels: user-experience
> Fix For: 0.11.0.0
>
>
> See discusion: https://github.com/apache/kafka/pull/2815
> We will need some rewrite on {{StreamPartitionsAssignor}} and to add much 
> more test for all kind of corner cases, including pattern subscriptions.



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

Re: KIP-162: Enable topic deletion by default

[Bill Bejeck]

+1

On Fri, May 26, 2017 at 10:55 AM, Gwen Shapira  wrote:

> This was a discussion, not a vote (sorry for mangling the title), but
> thanks for all the +1 anyway.
>
> Regarding Ismael's feedback:
> The current server.properties includes the following:
> # Switch to enable topic deletion or not, default value is false
> #delete.topic.enable=true
>
> We can't leave it as is, obviously - since the KIP invalidates the
> comment.  Lets just remove those two lines?
>
> Note that all our proposed changes may break few community puppet/docker
> scripts that use these lines for "sed" that enables topic deletion.
>
> Gwen
>
> On Fri, May 26, 2017 at 5:41 PM Tom Crayford  wrote:
>
> > +1 (non-binding)
> >
> > On Fri, May 26, 2017 at 3:38 PM, Damian Guy 
> wrote:
> >
> > > +1
> > > Also agree with what Ismael said.
> > >
> > > On Fri, 26 May 2017 at 15:26 Ismael Juma  wrote:
> > >
> > > > Thanks for the KIP, sounds good to me. One comment: not sure we need
> to
> > > add
> > > > the config to server.properties. Do we expect people to change this
> > > > default?
> > > >
> > > > On Fri, May 26, 2017 at 3:03 PM, Gwen Shapira 
> > wrote:
> > > >
> > > > > Hi Kafka developers, users and friends,
> > > > >
> > > > > I've added a KIP to improve our out-of-the-box usability a bit:
> > > > > KIP-162: Enable topic deletion by default:
> > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+
> > > > > Enable+topic+deletion+by+default
> > > > >
> > > > > Pretty simple :) Discussion and feedback are welcome.
> > > > >
> > > > > Gwen
> > > > >
> > > >
> > >
> >
>

[jira] [Commented] (KAFKA-5054) ChangeLoggingKeyValueByteStore delete and putIfAbsent should be synchronized

[Eno Thereska (JIRA)]

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

Eno Thereska commented on KAFKA-5054:
-

[~damianguy] is this still for 0.11?

> ChangeLoggingKeyValueByteStore delete and putIfAbsent should be synchronized
> 
>
> Key: KAFKA-5054
> URL: https://issues.apache.org/jira/browse/KAFKA-5054
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Damian Guy
>Assignee: Damian Guy
> Fix For: 0.11.0.0
>
>
> {{putIfAbsent}} and {{delete}} should be synchronized as they involve at 
> least 2 operations on the underlying store and may result in inconsistent 
> results if someone were to query via IQ



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

[jira] [Updated] (KAFKA-5301) Improve exception handling on consumer path

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-5301:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve exception handling on consumer path
> ---
>
> Key: KAFKA-5301
> URL: https://issues.apache.org/jira/browse/KAFKA-5301
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Eno Thereska
>Assignee: Eno Thereska
> Fix For: 0.11.1.0
>
>
> Used in StreamsThread.java, mostly to .poll() but also to restore data.
> Used in StreamsTask.java, mostly to .pause(), .resume()
> All exceptions here are currently caught all the way up to the main running 
> loop in a broad catch(Exception e) statement in StreamThread.run().
> One main concern on the consumer path is handling deserialization errors that 
> happen before streams has even had a chance to look at the data: 
> https://issues.apache.org/jira/browse/KAFKA-5157  



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

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Eno Thereska]

See latest reply to Jan's note. I think I unnecessarily broadened the scope of 
this KIP to the point where it sounded like it handles all sorts of exceptions. 
The scope should be strictly limited to "poison pill" records for now. Will 
update KIP, 

Thanks
Eno
> On 26 May 2017, at 16:16, Matthias J. Sax  wrote:
> 
> "bad" for this case would mean, that we got an
> `DeserializationException`. I am not sure if any other processing error
> should be covered?
> 
> @Eno: this raises one one question. Might it be better to allow for two
> handlers instead of one? One for deserialization exception and one for
> all other exceptions from user code?
> 
> Just a thought.
> 
> 
> -Matthias
> 
> On 5/26/17 7:49 AM, Jim Jagielski wrote:
>> 
>>> On May 26, 2017, at 5:13 AM, Eno Thereska  wrote:
>>> 
>>> 
 
 
 With regard to `DeserializationException`, do you thing it might make
 sense to have a "dead letter queue" as a feature to provide out-of-the-box?
>>> 
>>> We could provide a special topic where bad messages go to, and then we'd 
>>> have to add a config option for the user to provide a topic. Is that what 
>>> you're thinking?
>>> 
>> 
>> For various definitions of "bad"??
>> 
> 

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Eno Thereska]

Hi Jan,

You're right. I think I got carried away and broadened the scope of this KIP 
beyond it's original purpose. This handler will only be there for 
deserialization errors, i.e., "poison pills" and is not intended to be a 
catch-all handler for all sorts of other problems (e.g., NPE exception in user 
code). Deserialization erros can happen either when polling or when 
deserialising from a state store. So that narrows down the scope of the KIP, 
will update it.

Thanks
Eno

> On 26 May 2017, at 11:31, Jan Filipiak  wrote:
> 
> Hi
> 
> unfortunatly no. Think about "caching" these records popping outta there or 
> multiple step Tasks (join,aggregate,repartiton all in one go) last 
> repartitioner might throw cause it cant determine the partition only because 
> a get on the join store cause a flush through the aggregates. This has 
> nothing todo with a ConsumerRecord at all. Especially not the one we most 
> recently processed.
> 
> To be completly honest. All but grining to a hold is not appealing to me at 
> all. Sure maybe lagmonitoring will call me on Sunday but I can at least be 
> confident its working the rest of the time.
> 
> Best Jan
> 
> PS.:
> 
> Hope you get my point. I am mostly complaing about
> 
> |public| |interface| |RecordExceptionHandler {|
> |||/**|
> |||* Inspect a record and the exception received|
> |||*/|
> |||HandlerResponse handle(that guy here >>>   ConsumerRecord<||byte||[], 
> ||byte||[]> record, Exception exception);|
> |}|
> ||
> |public| |enum| |HandlerResponse {|
> |||/* continue with processing */|
> |||CONTINUE(||1||), |
> |||/* fail the processing and stop */|
> |||FAIL(||2||);|
> |}|
> 
> 
> 
> On 26.05.2017 11:18, Eno Thereska wrote:
>> Thanks Jan,
>> 
>> The record passed to the handler will always be the problematic record. 
>> There are 2 cases/types of exceptions for the purposes of this KIP: 1) any 
>> exception during deserialization. The bad record + the exception (i.e. 
>> DeserializeException) will be passed to the handler. The handler will be 
>> able to tell this was a deserialization error.
>> 2) any exception during processing of this record. So whenever a processor 
>> gets the record (after some caching, etc) it starts to process it, then it 
>> fails, then it will call the handler with this record.
>> 
>> Does that match your thinking?
>> 
>> Thanks,
>> Eno
>> 
>> 
>>> On 26 May 2017, at 09:51, Jan Filipiak  wrote:
>>> 
>>> Hi,
>>> 
>>> quick question: From the KIP it doesn't quite makes sense to me how that 
>>> fits with caching.
>>> With caching the consumer record might not be at all related to some 
>>> processor throwing while processing.
>>> 
>>> would it not make more sense to get the ProcessorName + object object for 
>>> processing and
>>> statestore or topic name + byte[] byte[]  for serializers? maybe passing in 
>>> the used serdes?
>>> 
>>> Best Jan
>>> 
>>> 
>>> 
>>> On 25.05.2017 11:47, Eno Thereska wrote:
 Hi there,
 
 I’ve added a KIP on improving exception handling in streams:
 KIP-161: streams record processing exception handlers. 
 https://cwiki.apache.org/confluence/display/KAFKA/KIP-161%3A+streams+record+processing+exception+handlers
  
 
 
 Discussion and feedback is welcome, thank you.
 Eno
> 

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Matthias J. Sax]

"bad" for this case would mean, that we got an
`DeserializationException`. I am not sure if any other processing error
should be covered?

@Eno: this raises one one question. Might it be better to allow for two
handlers instead of one? One for deserialization exception and one for
all other exceptions from user code?

Just a thought.


-Matthias

On 5/26/17 7:49 AM, Jim Jagielski wrote:
> 
>> On May 26, 2017, at 5:13 AM, Eno Thereska  wrote:
>>
>>
>>>
>>>
>>> With regard to `DeserializationException`, do you thing it might make
>>> sense to have a "dead letter queue" as a feature to provide out-of-the-box?
>>
>> We could provide a special topic where bad messages go to, and then we'd 
>> have to add a config option for the user to provide a topic. Is that what 
>> you're thinking?
>>
> 
> For various definitions of "bad"??
> 



signature.asc
Description: OpenPGP digital signature

Re: KIP-162: Enable topic deletion by default

[Gwen Shapira]

This was a discussion, not a vote (sorry for mangling the title), but
thanks for all the +1 anyway.

Regarding Ismael's feedback:
The current server.properties includes the following:
# Switch to enable topic deletion or not, default value is false
#delete.topic.enable=true

We can't leave it as is, obviously - since the KIP invalidates the
comment.  Lets just remove those two lines?

Note that all our proposed changes may break few community puppet/docker
scripts that use these lines for "sed" that enables topic deletion.

Gwen

On Fri, May 26, 2017 at 5:41 PM Tom Crayford  wrote:

> +1 (non-binding)
>
> On Fri, May 26, 2017 at 3:38 PM, Damian Guy  wrote:
>
> > +1
> > Also agree with what Ismael said.
> >
> > On Fri, 26 May 2017 at 15:26 Ismael Juma  wrote:
> >
> > > Thanks for the KIP, sounds good to me. One comment: not sure we need to
> > add
> > > the config to server.properties. Do we expect people to change this
> > > default?
> > >
> > > On Fri, May 26, 2017 at 3:03 PM, Gwen Shapira 
> wrote:
> > >
> > > > Hi Kafka developers, users and friends,
> > > >
> > > > I've added a KIP to improve our out-of-the-box usability a bit:
> > > > KIP-162: Enable topic deletion by default:
> > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+
> > > > Enable+topic+deletion+by+default
> > > >
> > > > Pretty simple :) Discussion and feedback are welcome.
> > > >
> > > > Gwen
> > > >
> > >
> >
>

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Jim Jagielski]

> On May 26, 2017, at 5:13 AM, Eno Thereska  wrote:
> 
> 
>> 
>> 
>> With regard to `DeserializationException`, do you thing it might make
>> sense to have a "dead letter queue" as a feature to provide out-of-the-box?
> 
> We could provide a special topic where bad messages go to, and then we'd have 
> to add a config option for the user to provide a topic. Is that what you're 
> thinking?
> 

For various definitions of "bad"??

Re: KIP-162: Enable topic deletion by default

[Tom Crayford]

+1 (non-binding)

On Fri, May 26, 2017 at 3:38 PM, Damian Guy  wrote:

> +1
> Also agree with what Ismael said.
>
> On Fri, 26 May 2017 at 15:26 Ismael Juma  wrote:
>
> > Thanks for the KIP, sounds good to me. One comment: not sure we need to
> add
> > the config to server.properties. Do we expect people to change this
> > default?
> >
> > On Fri, May 26, 2017 at 3:03 PM, Gwen Shapira  wrote:
> >
> > > Hi Kafka developers, users and friends,
> > >
> > > I've added a KIP to improve our out-of-the-box usability a bit:
> > > KIP-162: Enable topic deletion by default:
> > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+
> > > Enable+topic+deletion+by+default
> > >
> > > Pretty simple :) Discussion and feedback are welcome.
> > >
> > > Gwen
> > >
> >
>

Re: KIP-162: Enable topic deletion by default

[Damian Guy]

+1
Also agree with what Ismael said.

On Fri, 26 May 2017 at 15:26 Ismael Juma  wrote:

> Thanks for the KIP, sounds good to me. One comment: not sure we need to add
> the config to server.properties. Do we expect people to change this
> default?
>
> On Fri, May 26, 2017 at 3:03 PM, Gwen Shapira  wrote:
>
> > Hi Kafka developers, users and friends,
> >
> > I've added a KIP to improve our out-of-the-box usability a bit:
> > KIP-162: Enable topic deletion by default:
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+
> > Enable+topic+deletion+by+default
> >
> > Pretty simple :) Discussion and feedback are welcome.
> >
> > Gwen
> >
>

[GitHub] kafka pull request #3155: Fix KAKFA-5334.

[twbecker]

GitHub user twbecker opened a pull request:

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

Fix KAKFA-5334.

Handle` rocksdb.config.setter` being set as a class name or class
instance.

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

$ git pull https://github.com/twbecker/kafka KAFKA-5334

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

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

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

This closes #3155


commit 08b837410b5a2fbdf65727f46c44d1a008897708
Author: Tommy Becker 
Date:   2017-05-26T14:25:31Z

Fix KAKFA-5334.

Handle rocksdb.config.setter being set as a class name or class
instance.




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

Re: [DISCUSS]: KIP-161: streams record processing exception handlers

[Matthias J. Sax]

About `LogAndThresholdExceptionHandler`:

If the handler needs to keep track of number of failed messages, than it
becomes stateful -- not sure if we should do that. But maybe we can
introduce 2 metrics (might be an interesting metric to report to the
user anyway) and allow programmatic access to this metric in the
handler. This might also work better, as we should use "windowed"
failure rate anyway -- an absolute count of failed messages from
beginning of time would not work very well IMHO.

About "dead letter queue":

Yes, that what my thought -- we provide an out-of-the-box handler that
writes to a topic (thus, this part is covered by setting the
corresponding handler in the config). Thus, not sure if we need a config
for the topic name as we could exploit `Configurable` or we just use a
fixed name like "-dead-letters" (or some better name).
Not sure how important it is to configure the name (the less config the
better -- otherwise it's getting to hard to see what is
important/relevant and what not)


-Matthias


On 5/26/17 2:13 AM, Eno Thereska wrote:
> Replying to Avi's and Matthias' questions in one go inline:
> 
>> On 25 May 2017, at 19:27, Matthias J. Sax  wrote:
>>
>> Thanks for the KIP Eno!
>>
>> Couple of comments:
>>
>> I think we don't need `RecordContext` in `RecordExceptionHandler#handle`
>> because the `ConsumerRecord` provides all this information anyway.
> 
> Good point, fixed.
> 
>>
>> Why we introduce `ExceptionType` and not just hand in the actual exception?
>>
> 
> Ok, changed.
> 
>> As return type of `handle()` is void, how would the handler fail? By
>> throwing an exception? Maybe it would be better to add a proper return
>> type from the beginning on -- this might also make backward
>> compatibility easier later on.
> 
> Added two options to the KIP, see what you think (Continue/Fail). If we add 
> Retry later on, not sure if that'd trigger a KIP.
> 
>>
>> Question about `LogAndThresholdExceptionHandler` -- how would we be able
>> to track this?
> 
> I'm thinking StreamsThread instantiates one such handler that keeps a count 
> of failed messages. The handler will need to know the total of successful 
> messages too though, and come to think of it some more, we don't have a 
> metric that keeps that. We can either add that metric (in JMX) or 
> alternatively change the threshold handler to respond to number of failures, 
> instead of %. Preferences?
>>
>>
>> With regard to `DeserializationException`, do you thing it might make
>> sense to have a "dead letter queue" as a feature to provide out-of-the-box?
> 
> We could provide a special topic where bad messages go to, and then we'd have 
> to add a config option for the user to provide a topic. Is that what you're 
> thinking?
> 
> Eno
> 
> 
>>
>>
>> -Matthias
>>
>> On 5/25/17 2:47 AM, Eno Thereska wrote:
>>> Hi there,
>>>
>>> I’ve added a KIP on improving exception handling in streams:
>>> KIP-161: streams record processing exception handlers. 
>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-161%3A+streams+record+processing+exception+handlers
>>>  
>>> 
>>>
>>> Discussion and feedback is welcome, thank you.
>>> Eno
>>>
>>
> 



signature.asc
Description: OpenPGP digital signature

Re: KIP-162: Enable topic deletion by default

[Ismael Juma]

Thanks for the KIP, sounds good to me. One comment: not sure we need to add
the config to server.properties. Do we expect people to change this default?

On Fri, May 26, 2017 at 3:03 PM, Gwen Shapira  wrote:

> Hi Kafka developers, users and friends,
>
> I've added a KIP to improve our out-of-the-box usability a bit:
> KIP-162: Enable topic deletion by default:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+
> Enable+topic+deletion+by+default
>
> Pretty simple :) Discussion and feedback are welcome.
>
> Gwen
>

[jira] [Created] (KAFKA-5334) rocksdb.config.setter must be a class instance, not a class name

[Tommy Becker (JIRA)]

Tommy Becker created KAFKA-5334:
---

 Summary: rocksdb.config.setter must be a class instance, not a 
class name
 Key: KAFKA-5334
 URL: https://issues.apache.org/jira/browse/KAFKA-5334
 Project: Kafka
  Issue Type: Bug
  Components: streams
Affects Versions: 0.10.2.1
Reporter: Tommy Becker
Assignee: Tommy Becker
Priority: Minor


Unlike other config properties that are classes, {{rocksdb.config.setter}} 
cannot be a class _name_, it must be a class _instance_. This is because the 
raw config Map gets passed to RocksDBStore instead of the typed StreamsConfig, 
which is where the String -> Class conversion is handled. This means the config 
setter cannot be set via a config file, which is pretty limiting.



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

Re: KIP-162: Enable topic deletion by default

[Jorge Esteban Quilcate Otoya]

+1

El vie., 26 may. 2017 a las 16:14, Matthias J. Sax ()
escribió:

> +1
>
> On 5/26/17 7:03 AM, Gwen Shapira wrote:
> > Hi Kafka developers, users and friends,
> >
> > I've added a KIP to improve our out-of-the-box usability a bit:
> > KIP-162: Enable topic deletion by default:
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+Enable+topic+deletion+by+default
> >
> > Pretty simple :) Discussion and feedback are welcome.
> >
> > Gwen
> >
>
>

Re: KIP-162: Enable topic deletion by default

[Matthias J. Sax]

+1

On 5/26/17 7:03 AM, Gwen Shapira wrote:
> Hi Kafka developers, users and friends,
> 
> I've added a KIP to improve our out-of-the-box usability a bit:
> KIP-162: Enable topic deletion by default:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+Enable+topic+deletion+by+default
> 
> Pretty simple :) Discussion and feedback are welcome.
> 
> Gwen
> 



signature.asc
Description: OpenPGP digital signature

[jira] [Updated] (KAFKA-5226) NullPointerException (NPE) in SourceNodeRecordDeserializer.deserialize

[Bill Bejeck (JIRA)]

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

Bill Bejeck updated KAFKA-5226:
---
Fix Version/s: 0.11.0.0

> NullPointerException (NPE) in SourceNodeRecordDeserializer.deserialize
> --
>
> Key: KAFKA-5226
> URL: https://issues.apache.org/jira/browse/KAFKA-5226
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.1
> Environment: 64-bit Amazon Linux, JDK8
>Reporter: Ian Springer
>Assignee: Bill Bejeck
> Fix For: 0.11.0.0
>
> Attachments: kafka.log, streamsAppOne.log, streamsAppThree.log, 
> streamsAppTwo.log
>
>
> I saw the following NPE in our Kafka Streams app, which has 3 nodes running 
> on 3 separate machines.. Out of hundreds of messages processed, the NPE only 
> occurred twice. I are not sure of the cause, so I am unable to reproduce it. 
> I'm hoping the Kafka Streams team can guess the cause based on the stack 
> trace. If I can provide any additional details about our app, please let me 
> know.
>  
> {code}
> INFO  2017-05-10 02:58:26,021 org.apache.kafka.common.utils.AppInfoParser  
> Kafka version : 0.10.2.1
> INFO  2017-05-10 02:58:26,021 org.apache.kafka.common.utils.AppInfoParser  
> Kafka commitId : e89bffd6b2eff799
> INFO  2017-05-10 02:58:26,031 o.s.context.support.DefaultLifecycleProcessor  
> Starting beans in phase 0
> INFO  2017-05-10 02:58:26,075 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from CREATED to RUNNING.
> INFO  2017-05-10 02:58:26,075 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] Started 
> Kafka Stream process
> INFO  2017-05-10 02:58:26,086 o.a.k.c.consumer.internals.AbstractCoordinator  
> Discovered coordinator p1kaf1.prod.apptegic.com:9092 (id: 2147482646 rack: 
> null) for group evergage-app.
> INFO  2017-05-10 02:58:26,126 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partitions [] for group evergage-app
> INFO  2017-05-10 02:58:26,126 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from RUNNING to REBALANCING.
> INFO  2017-05-10 02:58:26,127 o.a.k.c.consumer.internals.AbstractCoordinator  
> (Re-)joining group evergage-app
> INFO  2017-05-10 02:58:27,712 o.a.k.c.consumer.internals.AbstractCoordinator  
> Successfully joined group evergage-app with generation 18
> INFO  2017-05-10 02:58:27,716 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Setting newly assigned partitions [us.app.Trigger-0] for group evergage-app
> INFO  2017-05-10 02:58:27,716 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to REBALANCING.
> INFO  2017-05-10 02:58:27,729 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> state stores
> INFO  2017-05-10 02:58:27,731 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> processor nodes of the topology
> INFO  2017-05-10 02:58:27,742 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to RUNNING.
> [14 hours pass...]
> INFO  2017-05-10 16:21:27,476 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partitions [us.app.Trigger-0] for group 
> evergage-app
> INFO  2017-05-10 16:21:27,477 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from RUNNING to REBALANCING.
> INFO  2017-05-10 16:21:27,482 o.a.k.c.consumer.internals.AbstractCoordinator  
> (Re-)joining group evergage-app
> INFO  2017-05-10 16:21:27,489 o.a.k.c.consumer.internals.AbstractCoordinator  
> Successfully joined group evergage-app with generation 19
> INFO  2017-05-10 16:21:27,489 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Setting newly assigned partitions [us.app.Trigger-0] for group evergage-app
> INFO  2017-05-10 16:21:27,489 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to REBALANCING.
> INFO  2017-05-10 16:21:27,489 
> o.a.kafka.streams.processor.internals.StreamTask  task [0_0] Initializing 
> processor nodes of the topology
> INFO  2017-05-10 16:21:27,493 org.apache.kafka.streams.KafkaStreams  
> stream-client [evergage-app-bd9c9868-4b9b-4d2e-850f-9b5bec1fc0a9] State 
> transition from REBALANCING to RUNNING.
> INFO  2017-05-10 16:21:30,584 o.a.k.c.consumer.internals.ConsumerCoordinator  
> Revoking previously assigned partitions [us.app.Trigger-

KIP-162: Enable topic deletion by default

[Gwen Shapira]

Hi Kafka developers, users and friends,

I've added a KIP to improve our out-of-the-box usability a bit:
KIP-162: Enable topic deletion by default:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-162+-+Enable+topic+deletion+by+default

Pretty simple :) Discussion and feedback are welcome.

Gwen

[jira] [Updated] (KAFKA-5219) Move transaction expiration logic and scheduling to the Transaction Manager

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-5219:
---
Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Move transaction expiration logic and scheduling to the Transaction Manager
> ---
>
> Key: KAFKA-5219
> URL: https://issues.apache.org/jira/browse/KAFKA-5219
> Project: Kafka
>  Issue Type: Sub-task
>  Components: core
>Reporter: Damian Guy
> Fix For: 0.11.1.0
>
>
> Presently the transaction expiration logic is spread between the 
> {{TransactionStateManager}} and the {{TransactionCoordinator}}. It would be 
> best if it was all in the {{TransactionStateManager}}. This requires moving 
> the bulk of the commit/abort logic, too. 



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

[jira] [Updated] (KAFKA-5313) Improve exception handling on coordinator interactions

[Eno Thereska (JIRA)]

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

Eno Thereska updated KAFKA-5313:

Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> Improve exception handling on coordinator interactions
> --
>
> Key: KAFKA-5313
> URL: https://issues.apache.org/jira/browse/KAFKA-5313
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Eno Thereska
>Assignee: Eno Thereska
> Fix For: 0.11.1.0
>
>
> Exceptions during assignment of tasks are caught in ConsumerCoordinator.java 
> and streams becomes aware of them during the 
> StreamThread.onPartitionsAssigned() and StreamThread.onPartitionsRevoked() 
> methods. Eventually these exceptions go through StreamThread.pollRequests() 
> all the way up to StreamThread.runLoop() and will halt the stream thread that 
> is processing these exceptions. Other stream threads may continue processing, 
> however it is likely they will experience problems too soon after.
> Exceptions here include LockExceptions that are thrown if tasks cannot use a 
> particular directory due to previous tasks not releasing locks on them during 
> reassignment. 



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