{"action":"created","issue":{"url":"https://api.github.com/repos/vert-x3/vertx-kafka-client/issues/114","repository_url":"https://api.github.com/repos/vert-x3/vertx-kafka-client","labels_url":"https://api.github.com/repos/vert-x3/vertx-kafka-client/issues/114/labels{/name}","comments_url":"https://api.github.com/repos/vert-x3/vertx-kafka-client/issues/114/comments","events_url":"https://api.github.com/repos/vert-x3/vertx-kafka-client/issues/114/events","html_url":"https://github.com/vert-x3/vertx-kafka-client/issues/114","id":374344694,"node_id":"MDU6SXNzdWUzNzQzNDQ2OTQ=","number":114,"title":"NullPointerException if kafka producer has partitions that have no leader","user":{"login":"mouse256","id":1605980,"node_id":"MDQ6VXNlcjE2MDU5ODA=","avatar_url":"https://avatars2.githubusercontent.com/u/1605980?v=4","gravatar_id":"","url":"https://api.github.com/users/mouse256","html_url":"https://github.com/mouse256","followers_url":"https://api.github.com/users/mouse256/followers","following_url":"https://api.github.com/users/mouse256/following{/other_user}","gists_url":"https://api.github.com/users/mouse256/gists{/gist_id}","starred_url":"https://api.github.com/users/mouse256/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mouse256/subscriptions","organizations_url":"https://api.github.com/users/mouse256/orgs","repos_url":"https://api.github.com/users/mouse256/repos","events_url":"https://api.github.com/users/mouse256/events{/privacy}","received_events_url":"https://api.github.com/users/mouse256/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2018-10-26T11:35:34Z","updated_at":"2018-11-19T16:00:48Z","closed_at":null,"author_association":"NONE","body":"If kafka is in a unhealthy state and has partitions that currently don't have a leader, it will make the kafkaproducer crash with a NullPointerException.\r\nI believe those partitions should be ignored in that case.\r\n\r\nStacktrace that happens in that case:\r\n`ERROR i.v.c.i.ContextImpl - Unhandled exception {}\r\n java.lang.NullPointerException: null\r\n at io.vertx.kafka.client.common.impl.Helper.from(Helper.java:87) ~[vertx-kafka-client-3.5.2.jar:3.5.2]\r\n at io.vertx.kafka.client.producer.impl.KafkaProducerImpl.lambda$partitionsFor$7(KafkaProducerImpl.java:165) ~[vertx-kafka-client-3.5.2.jar:3.5.2]\r\n at io.vertx.core.impl.FutureImpl.setHandler(FutureImpl.java:79) ~[vertx-core-3.5.3.jar:3.5.3]\r\n at io.vertx.core.impl.ContextImpl.lambda$null$0(ContextImpl.java:289) ~[vertx-core-3.5.3.jar:3.5.3]\r\n at io.vertx.core.impl.ContextImpl.lambda$wrapTask$2(ContextImpl.java:339) ~[vertx-core-3.5.3.jar:3.5.3]\r\n at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163) [netty-common-4.1.19.Final.jar:4.1.19.Final]\r\n at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404) [netty-common-4.1.19.Final.jar:4.1.19.Final]\r\n at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463) [netty-transport-4.1.19.Final.jar:4.1.19.Final]\r\n at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886) [netty-common-4.1.19.Final.jar:4.1.19.Final]\r\n at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.19.Final.jar:4.1.19.Final]\r\n`"},"comment":{"url":"https://api.github.com/repos/vert-x3/vertx-kafka-client/issues/comments/439944241","html_url":"https://github.com/vert-x3/vertx-kafka-client/issues/114#issuecomment-439944241","issue_url":"https://api.github.com/repos/vert-x3/vertx-kafka-client/issues/114","id":439944241,"node_id":"MDEyOklzc3VlQ29tbWVudDQzOTk0NDI0MQ==","user":{"login":"anhldbk","id":3270746,"node_id":"MDQ6VXNlcjMyNzA3NDY=","avatar_url":"https://avatars2.githubusercontent.com/u/3270746?v=4","gravatar_id":"","url":"https://api.github.com/users/anhldbk","html_url":"https://github.com/anhldbk","followers_url":"https://api.github.com/users/anhldbk/followers","following_url":"https://api.github.com/users/anhldbk/following{/other_user}","gists_url":"https://api.github.com/users/anhldbk/gists{/gist_id}","starred_url":"https://api.github.com/users/anhldbk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/anhldbk/subscriptions","organizations_url":"https://api.github.com/users/anhldbk/orgs","repos_url":"https://api.github.com/users/anhldbk/repos","events_url":"https://api.github.com/users/anhldbk/events{/privacy}","received_events_url":"https://api.github.com/users/anhldbk/received_events","type":"User","site_admin":false},"created_at":"2018-11-19T16:00:48Z","updated_at":"2018-11-19T16:00:48Z","author_association":"NONE","body":"@ppatierno Following this \b[post](https://news.ycombinator.com/item?id=15604686), I think we should have checks in the [loop](https://github.com/vert-x3/vertx-kafka-client/blob/master/src/main/java/io/vertx/kafka/client/producer/impl/KafkaProducerImpl.java#L158-L171)\r\n\r\n```java\r\nfor (org.apache.kafka.common.PartitionInfo kafkaPartitionInfo: done.result()) {\r\n\r\n\tPartitionInfo partitionInfo = new PartitionInfo();\r\n\r\n\tpartitionInfo\r\n\t.setInSyncReplicas(\r\n\t\tStream.of(kafkaPartitionInfo.inSyncReplicas()).map(Helper::from).collect(Collectors.toList()))\r\n\t.setLeader(Helper.from(kafkaPartitionInfo.leader())) // -> WE MAY NEED TO CHECK IF THIS VALUE IS NULL\r\n\t.setPartition(kafkaPartitionInfo.partition())\r\n\t.setReplicas(\r\n\t\tStream.of(kafkaPartitionInfo.replicas()).map(Helper::from).collect(Collectors.toList()))\r\n\t.setTopic(kafkaPartitionInfo.topic());\r\n\r\n\tpartitions.add(partitionInfo);\r\n}\r\n\thandler.handle(Future.succeededFuture(partitions));\r\n} else {\r\n\thandler.handle(Future.failedFuture(done.cause()));\r\n}\r\n```\r\nWe also need to fix that logic in `KafkaConsumerImpl`.\r\n\r\nShould I make a PR with the approach above?"}}
{"actor":{"display_login":"anhldbk"}}