{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/17462","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/17462/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/17462/comments","events_url":"https://api.github.com/repos/apache/spark/issues/17462/events","html_url":"https://github.com/apache/spark/pull/17462","id":217713092,"number":17462,"title":"[SPARK-20050][DStream] Kafka 0.10 DirectStream doesn't commit last processed batch's offset when graceful shutdown","user":{"login":"sasakitoa","id":9620266,"avatar_url":"https://avatars3.githubusercontent.com/u/9620266?v=4","gravatar_id":"","url":"https://api.github.com/users/sasakitoa","html_url":"https://github.com/sasakitoa","followers_url":"https://api.github.com/users/sasakitoa/followers","following_url":"https://api.github.com/users/sasakitoa/following{/other_user}","gists_url":"https://api.github.com/users/sasakitoa/gists{/gist_id}","starred_url":"https://api.github.com/users/sasakitoa/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sasakitoa/subscriptions","organizations_url":"https://api.github.com/users/sasakitoa/orgs","repos_url":"https://api.github.com/users/sasakitoa/repos","events_url":"https://api.github.com/users/sasakitoa/events{/privacy}","received_events_url":"https://api.github.com/users/sasakitoa/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2017-03-28T22:36:03Z","updated_at":"2017-12-14T21:03:39Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/17462","html_url":"https://github.com/apache/spark/pull/17462","diff_url":"https://github.com/apache/spark/pull/17462.diff","patch_url":"https://github.com/apache/spark/pull/17462.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nWhen we use KafkaDirectStream(Kafka0.10) with `enable.auto.commit` is `false`, \r\nsome record's offsets do not commit to Kafka at graceful shutdown, such below.\r\n\r\nSample code\r\n\r\n```\r\nval kafkaParams = Map[String, Object](\"enable.auto.commit\" -> \"false\")\r\nval kafkaStream = KafkaUtils.createDirectStream[String, String](ssc, ... , kafkaParams.asScala)\r\n\r\nkafkaStream.map { input =>\r\n  \"key: \" + input.key.toString + \" value: \" + input.value.toString + \" offset: \" + input.offset.toString\r\n  }.foreachRDD { rdd =>\r\n    rdd.foreach { input =>\r\n    println(input)\r\n  }\r\n}\r\n\r\nkafkaStream.foreachRDD { rdd =>\r\n  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges\r\n  kafkaStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)\r\n}\r\n```\r\n\r\noutput at first time\r\n\r\n```\r\nkey: null value: 1 offset: 101452472\r\nkey: null value: 2 offset: 101452473\r\nkey: null value: 3 offset: 101452474\r\nkey: null value: 4 offset: 101452475\r\nkey: null value: 5 offset: 101452476\r\nkey: null value: 6 offset: 101452477\r\nkey: null value: 7 offset: 101452478\r\nkey: null value: 8 offset: 101452479\r\nkey: null value: 9 offset: 101452480  // this is a last record before shutdown Spark Streaming gracefully\r\n```\r\n\r\noutput at second time\r\n\r\n```\r\nkey: null value: 7 offset: 101452478   // duplicate\r\nkey: null value: 8 offset: 101452479   // duplicate\r\nkey: null value: 9 offset: 101452480   // duplicate\r\nkey: null value: 10 offset: 101452481\r\n```\r\n\r\nThis is because offset will commit at the beginning of each batches, and will not commit after all batches processed.\r\n\r\n\r\n\r\n## How was this patch tested?\r\n\r\nAdded tests `offset commit when graceful shtudown` in `DirectKafkaStreamSuite.scala`"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/351835029","html_url":"https://github.com/apache/spark/pull/17462#issuecomment-351835029","issue_url":"https://api.github.com/repos/apache/spark/issues/17462","id":351835029,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars3.githubusercontent.com/u/4031291?v=4","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2017-12-14T21:03:39Z","updated_at":"2017-12-14T21:03:39Z","author_association":"NONE","body":"Can one of the admins verify this patch?"}}
{"actor":{"display_login":"AmplabJenkins"}}