[jira] [Commented] (KAFKA-3539) KafkaProducer.send() may block even though it returns the Future
[ https://issues.apache.org/jira/browse/KAFKA-3539?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16921598#comment-16921598 ] alex gabriel commented on KAFKA-3539: - [~tu...@avast.com] [~stevenz3wu] I think that you can avoid additional ExecutorService creation to make #send fully non-blocking by specifying max.block.ms to 0. But you still need to catch delivery exception until metadata comes. In your current solutions(if I got it right) you still have the chance to lose all the events that were allocated inside your ExecutorService queue( that is not persistent) since you only add events to the persistence storage only after rejection exceptions. > KafkaProducer.send() may block even though it returns the Future > > > Key: KAFKA-3539 > URL: https://issues.apache.org/jira/browse/KAFKA-3539 > Project: Kafka > Issue Type: Bug > Components: producer >Reporter: Oleg Zhurakousky >Priority: Critical > Labels: needs-discussion, needs-kip > > You can get more details from the us...@kafka.apache.org by searching on the > thread with the subject "KafkaProducer block on send". > The bottom line is that method that returns Future must never block, since it > essentially violates the Future contract as it was specifically designed to > return immediately passing control back to the user to check for completion, > cancel etc. -- This message was sent by Atlassian Jira (v8.3.2#803003)
[jira] [Updated] (KAFKA-8206) A consumer can't discover new group coordinator when the cluster was partly restarted
[
https://issues.apache.org/jira/browse/KAFKA-8206?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
alex gabriel updated KAFKA-8206:
Description:
*A consumer can't discover new group coordinator when the cluster was partly
restarted*
Preconditions:
I use Kafka server and Java kafka-client lib 2.2 version
I have 2 Kafka nodes running localy (localhost:9092, localhost:9093) and 1
ZK(localhost:2181)
I have replication factor 2 for the all my topics and
'_unclean.leader.election.enable=true_' on both Kafka nodes.
Steps to reproduce:
1) Start 2nodes (localhost:9092/localhost:9093)
2) Start consumer with 'bootstrap.servers=localhost:9092,localhost:9093'
{noformat}
// discovered group coordinator (0-node)
2019-04-09 16:23:18,963 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9092 (id: 2147483647 rack: null)>
...metadatacache is updated (2 nodes in the cluster list)
2019-04-09 16:23:18,928 DEBUG
[org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate] -
[Consumer clientId=events-consumer0, groupId=events-group-gabriel] Sending
metadata request (type=MetadataRequest, topics=) to node localhost:9092
(id: -1 rack: null)>
2019-04-09 16:23:18,940 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 2 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null),
localhost:9093 (id: 1 rack: null)], partitions = [], controller =
localhost:9092 (id: 0 rack: null))}>
{noformat}
3) Shutdown 1-node (localhost:9093)
{noformat}
// metadata was updated to the 4 version (but for some reasons it still had 2
alive nodes inside cluster)
2019-04-09 16:23:46,981 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 4 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9093 (id: 1 rack: null),
localhost:9092 (id: 0 rack: null)], partitions = [Partition(topic =
events-sorted, partition = 1, leader = 0, replicas = [0,1], isr = [0,1],
offlineReplicas = []), Partition(topic = events-sorted, partition = 0, leader =
0, replicas = [0,1], isr = [0,1], offlineReplicas = [])], controller =
localhost:9092 (id: 0 rack: null))}>
//consumers thinks that node-1 is still alive and try to send coordinator
lookup to it but failed
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9093 (id: 2147483646 rack: null)>
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.markCoordinatorUnknown]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Group
coordinator localhost:9093 (id: 2147483646 rack: null) is unavailable or
invalid, will attempt rediscovery>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.NetworkClient.handleDisconnections] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Node 1 disconnected.>
2019-04-09 16:24:01,117 WARN
[org.apache.kafka.clients.NetworkClient.processDisconnection] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Connection to node 1
(localhost:9093) could not be established. Broker may not be available.>
// refreshing metadata again
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Cancelled
request with header RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=2,
clientId=events-consumer0, correlationId=112) due to node 1 being disconnected>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Coordinator discovery failed, refreshing metadata>
// metadata was updated to the 5 version where cluster had only 0-node
localhost:9092 as expected.
2019-04-09 16:24:01,131 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 5 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null)], partitions
= [Partition(topic = events-sorted, partition = 1, leader = 0, replicas =
[0,1], isr = [0], offlineReplicas = [1]), Partition(topic = events-sorted,
partition = 0, leader = 0, replicas = [0,1], isr = [0], offlineReplicas =
[1])], controller = localhost:9092 (id: 0 rack: null))}>
// 0-node discovered as coordinator
2019-04-09 16:24:01,132 INFO
[jira] [Updated] (KAFKA-8206) A consumer can't discover new group coordinator when the cluster was partly restarted
[
https://issues.apache.org/jira/browse/KAFKA-8206?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
alex gabriel updated KAFKA-8206:
Description:
*A consumer can't discover new group coordinator when the cluster was partly
restarted*
Preconditions:
I use Kafka server and Java kafka-client lib 2.2 version
I have 2 Kafka nodes running localy (localhost:9092, localhost:9093) and 1
ZK(localhost:2181)
I have replication factor 2 for the all my topics and
'_unclean.leader.election.enable=true_' on both Kafka nodes.
Steps to reproduce:
1) Start 2nodes (localhost:9092/localhost:9093)
2) Start consumer with 'bootstrap.servers=localhost:9092,localhost:9093'
{noformat}
// discovered group coordinator (0-node)
2019-04-09 16:23:18,963 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9092 (id: 2147483647 rack: null)>
...metadatacache is updated (2 nodes in the cluster list)
2019-04-09 16:23:18,928 DEBUG
[org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate] -
[Consumer clientId=events-consumer0, groupId=events-group-gabriel] Sending
metadata request (type=MetadataRequest, topics=) to node localhost:9092
(id: -1 rack: null)>
2019-04-09 16:23:18,940 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 2 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null),
localhost:9093 (id: 1 rack: null)], partitions = [], controller =
localhost:9092 (id: 0 rack: null))}>
{noformat}
3) Shutdown 1-node (localhost:9093)
{noformat}
// metadata was updated to the 4 version (but for some reasons it still had 2
alive nodes inside cluster)
2019-04-09 16:23:46,981 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 4 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9093 (id: 1 rack: null),
localhost:9092 (id: 0 rack: null)], partitions = [Partition(topic =
events-sorted, partition = 1, leader = 0, replicas = [0,1], isr = [0,1],
offlineReplicas = []), Partition(topic = events-sorted, partition = 0, leader =
0, replicas = [0,1], isr = [0,1], offlineReplicas = [])], controller =
localhost:9092 (id: 0 rack: null))}>
//consumers thinks that node-1 is still alive and try to send coordinator
lookup to it but failed
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9093 (id: 2147483646 rack: null)>
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.markCoordinatorUnknown]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Group
coordinator localhost:9093 (id: 2147483646 rack: null) is unavailable or
invalid, will attempt rediscovery>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.NetworkClient.handleDisconnections] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Node 1 disconnected.>
2019-04-09 16:24:01,117 WARN
[org.apache.kafka.clients.NetworkClient.processDisconnection] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Connection to node 1
(localhost:9093) could not be established. Broker may not be available.>
// refreshing metadata again
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Cancelled
request with header RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=2,
clientId=events-consumer0, correlationId=112) due to node 1 being disconnected>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Coordinator discovery failed, refreshing metadata>
// metadata was updated to the 5 version where cluster had only 0-node
localhost:9092 as expected.
2019-04-09 16:24:01,131 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 5 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null)], partitions
= [Partition(topic = events-sorted, partition = 1, leader = 0, replicas =
[0,1], isr = [0], offlineReplicas = [1]), Partition(topic = events-sorted,
partition = 0, leader = 0, replicas = [0,1], isr = [0], offlineReplicas =
[1])], controller = localhost:9092 (id: 0 rack: null))}>
// 0-node discovered as coordinator
2019-04-09 16:24:01,132 INFO
[jira] [Updated] (KAFKA-8206) A consumer can't discover new group coordinator when the cluster was partly restarted
[
https://issues.apache.org/jira/browse/KAFKA-8206?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
alex gabriel updated KAFKA-8206:
Description:
*A consumer can't discover new group coordinator when the cluster was partly
restarted*
Preconditions:
I use Kafka server and Java kafka-client lib 2.2 version
I have 2 Kafka nodes running localy (localhost:9092, localhost:9093) and 1
ZK(localhost:2181)
I have replication factor 2 for the all my topics and
'_unclean.leader.election.enable=true_' on both Kafka nodes.
Steps to reproduce:
1) Start 2nodes (localhost:9092/localhost:9093)
2) Start consumer with 'bootstrap.servers=localhost:9092,localhost:9093'
{noformat}
// discovered group coordinator (0-node)
2019-04-09 16:23:18,963 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9092 (id: 2147483647 rack: null)>
...metadatacache is updated (2 nodes in the cluster list)
2019-04-09 16:23:18,928 DEBUG
[org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate] -
[Consumer clientId=events-consumer0, groupId=events-group-gabriel] Sending
metadata request (type=MetadataRequest, topics=) to node localhost:9092
(id: -1 rack: null)>
2019-04-09 16:23:18,940 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 2 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null),
localhost:9093 (id: 1 rack: null)], partitions = [], controller =
localhost:9092 (id: 0 rack: null))}>
{noformat}
3) Shutdown 1-node (localhost:9093)
{noformat}
// metadata was updated to the 4 version (but for some reasons it still had 2
alive nodes inside cluster)
2019-04-09 16:23:46,981 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 4 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9093 (id: 1 rack: null),
localhost:9092 (id: 0 rack: null)], partitions = [Partition(topic =
events-sorted, partition = 1, leader = 0, replicas = [0,1], isr = [0,1],
offlineReplicas = []), Partition(topic = events-sorted, partition = 0, leader =
0, replicas = [0,1], isr = [0,1], offlineReplicas = [])], controller =
localhost:9092 (id: 0 rack: null))}>
//consumers thinks that node-1 is still alive and try to send coordinator
lookup to it but failed
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9093 (id: 2147483646 rack: null)>
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.markCoordinatorUnknown]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Group
coordinator localhost:9093 (id: 2147483646 rack: null) is unavailable or
invalid, will attempt rediscovery>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.NetworkClient.handleDisconnections] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Node 1 disconnected.>
2019-04-09 16:24:01,117 WARN
[org.apache.kafka.clients.NetworkClient.processDisconnection] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Connection to node 1
(localhost:9093) could not be established. Broker may not be available.>
// refreshing metadata again
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Cancelled
request with header RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=2,
clientId=events-consumer0, correlationId=112) due to node 1 being disconnected>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Coordinator discovery failed, refreshing metadata>
// metadata was updated to the 5 version where cluster had only 0-node
localhost:9092 as expected.
2019-04-09 16:24:01,131 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 5 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null)], partitions
= [Partition(topic = events-sorted, partition = 1, leader = 0, replicas =
[0,1], isr = [0], offlineReplicas = [1]), Partition(topic = events-sorted,
partition = 0, leader = 0, replicas = [0,1], isr = [0], offlineReplicas =
[1])], controller = localhost:9092 (id: 0 rack: null))}>
// 0-node discovered as coordinator
2019-04-09 16:24:01,132 INFO
[jira] [Created] (KAFKA-8206) A consumer can't discover new group coordinator when the cluster was partly restarted
alex gabriel created KAFKA-8206:
---
Summary: A consumer can't discover new group coordinator when the
cluster was partly restarted
Key: KAFKA-8206
URL: https://issues.apache.org/jira/browse/KAFKA-8206
Project: Kafka
Issue Type: Bug
Affects Versions: 2.2.0, 2.0.0, 1.0.0
Reporter: alex gabriel
*A consumer can't discover new group coordinator when the cluster was partly
restarted*
Preconditions:
I use Kafka server and Java kafka-client lib 2.2 version
I have 2 Kafka nodes running localy (localhost:9092, localhost:9093) and 1
ZK(localhost:2181/localhost:2181)
I have replication factor 2 for the all my topics and
'_unclean.leader.election.enable=true_' on both Kafka nodes.
Steps to reproduce:
1) Start 2nodes (localhost:9092/localhost:9093)
2) Start consumer with 'bootstrap.servers=localhost:9092,localhost:9093'
{noformat}
// discovered group coordinator (0-node)
2019-04-09 16:23:18,963 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9092 (id: 2147483647 rack: null)>
...metadatacache is updated (2 nodes in the cluster list)
2019-04-09 16:23:18,928 DEBUG
[org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate] -
[Consumer clientId=events-consumer0, groupId=events-group-gabriel] Sending
metadata request (type=MetadataRequest, topics=) to node localhost:9092
(id: -1 rack: null)>
2019-04-09 16:23:18,940 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 2 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null),
localhost:9093 (id: 1 rack: null)], partitions = [], controller =
localhost:9092 (id: 0 rack: null))}>
{noformat}
3) Shutdown 1-node (localhost:9093)
{noformat}
// metadata was updated to the 4 version (but for some reasons it still had 2
alive nodes inside cluster)
2019-04-09 16:23:46,981 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 4 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9093 (id: 1 rack: null),
localhost:9092 (id: 0 rack: null)], partitions = [Partition(topic =
events-sorted, partition = 1, leader = 0, replicas = [0,1], isr = [0,1],
offlineReplicas = []), Partition(topic = events-sorted, partition = 0, leader =
0, replicas = [0,1], isr = [0,1], offlineReplicas = [])], controller =
localhost:9092 (id: 0 rack: null))}>
//consumers thinks that node-1 is still alive and try to send coordinator
lookup to it but failed
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator$FindCoordinatorResponseHandler.onSuccess]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Discovered group coordinator localhost:9093 (id: 2147483646 rack: null)>
2019-04-09 16:23:46,981 INFO
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.markCoordinatorUnknown]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Group
coordinator localhost:9093 (id: 2147483646 rack: null) is unavailable or
invalid, will attempt rediscovery>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.NetworkClient.handleDisconnections] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Node 1 disconnected.>
2019-04-09 16:24:01,117 WARN
[org.apache.kafka.clients.NetworkClient.processDisconnection] - [Consumer
clientId=events-consumer0, groupId=events-group-gabriel] Connection to node 1
(localhost:9093) could not be established. Broker may not be available.>
// refreshing metadata again
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel] Cancelled
request with header RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=2,
clientId=events-consumer0, correlationId=112) due to node 1 being disconnected>
2019-04-09 16:24:01,117 DEBUG
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady]
- [Consumer clientId=events-consumer0, groupId=events-group-gabriel]
Coordinator discovery failed, refreshing metadata>
// metadata was updated to the 5 version where cluster had only 0-node
localhost:9092 as expected.
2019-04-09 16:24:01,131 DEBUG [org.apache.kafka.clients.Metadata.update] -
Updated cluster metadata version 5 to MetadataCache{cluster=Cluster(id =
P3pz1xU0SjK-Dhy6h2G5YA, nodes = [localhost:9092 (id: 0 rack: null)], partitions
= [Partition(topic = events-sorted, partition = 1, leader = 0, replicas =
[0,1], isr = [0], offlineReplicas = [1]), Partition(topic = events-sorted,
partition = 0, leader = 0, replicas = [0,1], isr = [0], offlineReplicas =
