{"action":"created","comment":{"url":"https://api.github.com/repos/locationtech/geomesa/pulls/comments/31615224","id":31615224,"diff_hunk":"@@ -0,0 +1,419 @@\n+/*\n+ * Copyright 2015 Commonwealth Computer Research, Inc.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the License);\n+ * you may not use this file except in compliance with the License.\n+ * You may obtain a copy of the License at\n+ *\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an AS IS BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package org.locationtech.geomesa.kafka\n+\n+import java.util.Date\n+\n+import com.typesafe.scalalogging.slf4j.Logging\n+import com.vividsolutions.jts.index.quadtree.Quadtree\n+import org.geotools.data.store.ContentEntry\n+import org.geotools.data.{EmptyFeatureReader, Query}\n+import org.geotools.factory.CommonFactoryFinder\n+import org.geotools.feature.simple.{SimpleFeatureBuilder, SimpleFeatureTypeBuilder}\n+import org.joda.time.{Duration, Instant}\n+import org.locationtech.geomesa.filter._\n+import org.locationtech.geomesa.kafka.consumer.KafkaConsumerFactory\n+import org.locationtech.geomesa.kafka.consumer.KafkaStreamLike.KafkaStreamLikeIterator\n+import org.locationtech.geomesa.kafka.consumer.offsets.FindOffset\n+import org.locationtech.geomesa.utils.geotools.Conversions._\n+import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes.FR\n+import org.opengis.feature.simple.{SimpleFeature, SimpleFeatureType}\n+import org.opengis.filter._\n+import org.opengis.filter.expression.PropertyName\n+\n+import scala.collection.JavaConverters._\n+import scala.collection.mutable\n+\n+class ReplayKafkaConsumerFeatureSource(entry: ContentEntry,\n+                                       replaySFT: SimpleFeatureType,\n+                                       liveSFT: SimpleFeatureType,\n+                                       topic: String,\n+                                       kf: KafkaConsumerFactory,\n+                                       replayConfig: ReplayConfig,\n+                                       query: Query = null)\n+  extends KafkaConsumerFeatureSource(entry, replaySFT, query)\n+  with Logging {\n+\n+  import TimestampFilterSplit.split\n+\n+  // messages are stored as an array where the most recent is at index 0\n+  private[kafka] val messages: Array[GeoMessage] = readMessages()\n+\n+  override def getReaderForFilter(filter: Filter): FR = {\n+\n+    val reader: Option[FR] = if (messages.isEmpty) {\n+      // no data!\n+      None\n+    } else {\n+      split(filter).flatMap { tfs =>\n+        val time = tfs.ts\n+        val filter = tfs.filter.getOrElse(Filter.INCLUDE)\n+\n+        snapshot(time).map(_.getReaderForFilter(filter))\n+      }\n+    }\n+\n+    reader.getOrElse(new EmptyFeatureReader[SimpleFeatureType, SimpleFeature](replaySFT))\n+  }\n+\n+  private[kafka] def snapshot(time: Option[Long]): Option[ReplaySnapshotFeatureCache] = {\n+\n+    val (startTime, startIndex) = time\n+      .map(ts => (ts, indexAtTime(ts)))\n+      .getOrElse((replayConfig.end.getMillis, Some(0)))\n+\n+    startIndex.map { si =>\n+      val endTime = startTime - replayConfig.readBehind.getMillis\n+      snapshot(startTime, si, endTime)\n+    }.getOrElse(None)\n+  }\n+\n+  /** @return the index of the most recent [[GeoMessage]] at or before the given ``time``\n+    */\n+  private[kafka] def indexAtTime(time: Long): Option[Int] = {\n+\n+    if (replayConfig.isInWindow(time)) {\n+      // it doesn't matter what the message is, only the time\n+      val key: GeoMessage = new Clear(new Instant(time))\n+\n+      // reverse ordering for reverse ordered ``messages``\n+      val ordering = new Ordering[GeoMessage] {\n+        override def compare(x: GeoMessage, y: GeoMessage): Int = y.timestamp.compareTo(x.timestamp)\n+      }\n+\n+      var index = java.util.Arrays.binarySearch(messages, key, ordering)\n+\n+      if (index < 0) {\n+        // no message found at sought time\n+        index = -index - 1\n+      }\n+\n+      // there may be multiple messages at the same time so there may be messages before ``index``\n+      // that are at the given ``time``\n+      while (index > 0 && messages(index - 1).timestamp.getMillis <= time) index -= 1","path":"geomesa-kafka/geomesa-kafka-datastore/src/main/scala/org/locationtech/geomesa/kafka/ReplayKafkaConsumerFeatureSource.scala","position":106,"original_position":106,"commit_id":"5a866421a301c2cfe480376f240049333050d44b","original_commit_id":"5a866421a301c2cfe480376f240049333050d44b","user":{"login":"elahrvivaz","id":7715322,"avatar_url":"https://avatars.githubusercontent.com/u/7715322?v=3","gravatar_id":"","url":"https://api.github.com/users/elahrvivaz","html_url":"https://github.com/elahrvivaz","followers_url":"https://api.github.com/users/elahrvivaz/followers","following_url":"https://api.github.com/users/elahrvivaz/following{/other_user}","gists_url":"https://api.github.com/users/elahrvivaz/gists{/gist_id}","starred_url":"https://api.github.com/users/elahrvivaz/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/elahrvivaz/subscriptions","organizations_url":"https://api.github.com/users/elahrvivaz/orgs","repos_url":"https://api.github.com/users/elahrvivaz/repos","events_url":"https://api.github.com/users/elahrvivaz/events{/privacy}","received_events_url":"https://api.github.com/users/elahrvivaz/received_events","type":"User","site_admin":false},"body":"{ }","created_at":"2015-06-03T11:54:52Z","updated_at":"2015-06-03T11:54:52Z","html_url":"https://github.com/locationtech/geomesa/pull/596#discussion_r31615224","pull_request_url":"https://api.github.com/repos/locationtech/geomesa/pulls/596","_links":{"self":{"href":"https://api.github.com/repos/locationtech/geomesa/pulls/comments/31615224"},"html":{"href":"https://github.com/locationtech/geomesa/pull/596#discussion_r31615224"},"pull_request":{"href":"https://api.github.com/repos/locationtech/geomesa/pulls/596"}}},"pull_request":{"url":"https://api.github.com/repos/locationtech/geomesa/pulls/596","id":36648521,"html_url":"https://github.com/locationtech/geomesa/pull/596","diff_url":"https://github.com/locationtech/geomesa/pull/596.diff","patch_url":"https://github.com/locationtech/geomesa/pull/596.patch","issue_url":"https://api.github.com/repos/locationtech/geomesa/issues/596","number":596,"state":"open","locked":false,"title":"GEOMESA-780 Kafka Data Store Replay","user":{"login":"mmatz-ccri","id":11392977,"avatar_url":"https://avatars.githubusercontent.com/u/11392977?v=3","gravatar_id":"","url":"https://api.github.com/users/mmatz-ccri","html_url":"https://github.com/mmatz-ccri","followers_url":"https://api.github.com/users/mmatz-ccri/followers","following_url":"https://api.github.com/users/mmatz-ccri/following{/other_user}","gists_url":"https://api.github.com/users/mmatz-ccri/gists{/gist_id}","starred_url":"https://api.github.com/users/mmatz-ccri/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mmatz-ccri/subscriptions","organizations_url":"https://api.github.com/users/mmatz-ccri/orgs","repos_url":"https://api.github.com/users/mmatz-ccri/repos","events_url":"https://api.github.com/users/mmatz-ccri/events{/privacy}","received_events_url":"https://api.github.com/users/mmatz-ccri/received_events","type":"User","site_admin":false},"body":"* Added replay ability to Kafka Data Store.\r\n* GEOMESA-799 Kafka consumer with ability to start from arbitrary offset\r\n\r\nSigned-off-by: Michael Matz <mmatz@ccri.com>","created_at":"2015-06-01T16:58:41Z","updated_at":"2015-06-03T11:54:52Z","closed_at":null,"merged_at":null,"merge_commit_sha":"5aea159e64e6b1c830d09e1ab2536d04943a22a5","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/locationtech/geomesa/pulls/596/commits","review_comments_url":"https://api.github.com/repos/locationtech/geomesa/pulls/596/comments","review_comment_url":"https://api.github.com/repos/locationtech/geomesa/pulls/comments{/number}","comments_url":"https://api.github.com/repos/locationtech/geomesa/issues/596/comments","statuses_url":"https://api.github.com/repos/locationtech/geomesa/statuses/5a866421a301c2cfe480376f240049333050d44b","head":{"label":"ccri:fcr_kafka_replay","ref":"fcr_kafka_replay","sha":"5a866421a301c2cfe480376f240049333050d44b","user":{"login":"ccri","id":1089783,"avatar_url":"https://avatars.githubusercontent.com/u/1089783?v=3","gravatar_id":"","url":"https://api.github.com/users/ccri","html_url":"https://github.com/ccri","followers_url":"https://api.github.com/users/ccri/followers","following_url":"https://api.github.com/users/ccri/following{/other_user}","gists_url":"https://api.github.com/users/ccri/gists{/gist_id}","starred_url":"https://api.github.com/users/ccri/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ccri/subscriptions","organizations_url":"https://api.github.com/users/ccri/orgs","repos_url":"https://api.github.com/users/ccri/repos","events_url":"https://api.github.com/users/ccri/events{/privacy}","received_events_url":"https://api.github.com/users/ccri/received_events","type":"Organization","site_admin":false},"repo":{"id":16865543,"name":"geomesa","full_name":"ccri/geomesa","owner":{"login":"ccri","id":1089783,"avatar_url":"https://avatars.githubusercontent.com/u/1089783?v=3","gravatar_id":"","url":"https://api.github.com/users/ccri","html_url":"https://github.com/ccri","followers_url":"https://api.github.com/users/ccri/followers","following_url":"https://api.github.com/users/ccri/following{/other_user}","gists_url":"https://api.github.com/users/ccri/gists{/gist_id}","starred_url":"https://api.github.com/users/ccri/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ccri/subscriptions","organizations_url":"https://api.github.com/users/ccri/orgs","repos_url":"https://api.github.com/users/ccri/repos","events_url":"https://api.github.com/users/ccri/events{/privacy}","received_events_url":"https://api.github.com/users/ccri/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/ccri/geomesa","description":"docs","fork":true,"url":"https://api.github.com/repos/ccri/geomesa","forks_url":"https://api.github.com/repos/ccri/geomesa/forks","keys_url":"https://api.github.com/repos/ccri/geomesa/keys{/key_id}","collaborators_url":"https://api.github.com/repos/ccri/geomesa/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/ccri/geomesa/teams","hooks_url":"https://api.github.com/repos/ccri/geomesa/hooks","issue_events_url":"https://api.github.com/repos/ccri/geomesa/issues/events{/number}","events_url":"https://api.github.com/repos/ccri/geomesa/events","assignees_url":"https://api.github.com/repos/ccri/geomesa/assignees{/user}","branches_url":"https://api.github.com/repos/ccri/geomesa/branches{/branch}","tags_url":"https://api.github.com/repos/ccri/geomesa/tags","blobs_url":"https://api.github.com/repos/ccri/geomesa/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/ccri/geomesa/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/ccri/geomesa/git/refs{/sha}","trees_url":"https://api.github.com/repos/ccri/geomesa/git/trees{/sha}","statuses_url":"https://api.github.com/repos/ccri/geomesa/statuses/{sha}","languages_url":"https://api.github.com/repos/ccri/geomesa/languages","stargazers_url":"https://api.github.com/repos/ccri/geomesa/stargazers","contributors_url":"https://api.github.com/repos/ccri/geomesa/contributors","subscribers_url":"https://api.github.com/repos/ccri/geomesa/subscribers","subscription_url":"https://api.github.com/repos/ccri/geomesa/subscription","commits_url":"https://api.github.com/repos/ccri/geomesa/commits{/sha}","git_commits_url":"https://api.github.com/repos/ccri/geomesa/git/commits{/sha}","comments_url":"https://api.github.com/repos/ccri/geomesa/comments{/number}","issue_comment_url":"https://api.github.com/repos/ccri/geomesa/issues/comments{/number}","contents_url":"https://api.github.com/repos/ccri/geomesa/contents/{+path}","compare_url":"https://api.github.com/repos/ccri/geomesa/compare/{base}...{head}","merges_url":"https://api.github.com/repos/ccri/geomesa/merges","archive_url":"https://api.github.com/repos/ccri/geomesa/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/ccri/geomesa/downloads","issues_url":"https://api.github.com/repos/ccri/geomesa/issues{/number}","pulls_url":"https://api.github.com/repos/ccri/geomesa/pulls{/number}","milestones_url":"https://api.github.com/repos/ccri/geomesa/milestones{/number}","notifications_url":"https://api.github.com/repos/ccri/geomesa/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/ccri/geomesa/labels{/name}","releases_url":"https://api.github.com/repos/ccri/geomesa/releases{/id}","created_at":"2014-02-15T15:39:50Z","updated_at":"2015-03-26T17:21:45Z","pushed_at":"2015-06-01T17:19:23Z","git_url":"git://github.com/ccri/geomesa.git","ssh_url":"git@github.com:ccri/geomesa.git","clone_url":"https://github.com/ccri/geomesa.git","svn_url":"https://github.com/ccri/geomesa","homepage":"","size":9667,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"accumulo1.5.x/1.x"}},"base":{"label":"locationtech:master","ref":"master","sha":"0d607fc1479790a73930195f50505b3e8b582d55","user":{"login":"locationtech","id":1991411,"avatar_url":"https://avatars.githubusercontent.com/u/1991411?v=3","gravatar_id":"","url":"https://api.github.com/users/locationtech","html_url":"https://github.com/locationtech","followers_url":"https://api.github.com/users/locationtech/followers","following_url":"https://api.github.com/users/locationtech/following{/other_user}","gists_url":"https://api.github.com/users/locationtech/gists{/gist_id}","starred_url":"https://api.github.com/users/locationtech/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/locationtech/subscriptions","organizations_url":"https://api.github.com/users/locationtech/orgs","repos_url":"https://api.github.com/users/locationtech/repos","events_url":"https://api.github.com/users/locationtech/events{/privacy}","received_events_url":"https://api.github.com/users/locationtech/received_events","type":"Organization","site_admin":false},"repo":{"id":13226069,"name":"geomesa","full_name":"locationtech/geomesa","owner":{"login":"locationtech","id":1991411,"avatar_url":"https://avatars.githubusercontent.com/u/1991411?v=3","gravatar_id":"","url":"https://api.github.com/users/locationtech","html_url":"https://github.com/locationtech","followers_url":"https://api.github.com/users/locationtech/followers","following_url":"https://api.github.com/users/locationtech/following{/other_user}","gists_url":"https://api.github.com/users/locationtech/gists{/gist_id}","starred_url":"https://api.github.com/users/locationtech/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/locationtech/subscriptions","organizations_url":"https://api.github.com/users/locationtech/orgs","repos_url":"https://api.github.com/users/locationtech/repos","events_url":"https://api.github.com/users/locationtech/events{/privacy}","received_events_url":"https://api.github.com/users/locationtech/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/locationtech/geomesa","description":"","fork":false,"url":"https://api.github.com/repos/locationtech/geomesa","forks_url":"https://api.github.com/repos/locationtech/geomesa/forks","keys_url":"https://api.github.com/repos/locationtech/geomesa/keys{/key_id}","collaborators_url":"https://api.github.com/repos/locationtech/geomesa/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/locationtech/geomesa/teams","hooks_url":"https://api.github.com/repos/locationtech/geomesa/hooks","issue_events_url":"https://api.github.com/repos/locationtech/geomesa/issues/events{/number}","events_url":"https://api.github.com/repos/locationtech/geomesa/events","assignees_url":"https://api.github.com/repos/locationtech/geomesa/assignees{/user}","branches_url":"https://api.github.com/repos/locationtech/geomesa/branches{/branch}","tags_url":"https://api.github.com/repos/locationtech/geomesa/tags","blobs_url":"https://api.github.com/repos/locationtech/geomesa/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/locationtech/geomesa/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/locationtech/geomesa/git/refs{/sha}","trees_url":"https://api.github.com/repos/locationtech/geomesa/git/trees{/sha}","statuses_url":"https://api.github.com/repos/locationtech/geomesa/statuses/{sha}","languages_url":"https://api.github.com/repos/locationtech/geomesa/languages","stargazers_url":"https://api.github.com/repos/locationtech/geomesa/stargazers","contributors_url":"https://api.github.com/repos/locationtech/geomesa/contributors","subscribers_url":"https://api.github.com/repos/locationtech/geomesa/subscribers","subscription_url":"https://api.github.com/repos/locationtech/geomesa/subscription","commits_url":"https://api.github.com/repos/locationtech/geomesa/commits{/sha}","git_commits_url":"https://api.github.com/repos/locationtech/geomesa/git/commits{/sha}","comments_url":"https://api.github.com/repos/locationtech/geomesa/comments{/number}","issue_comment_url":"https://api.github.com/repos/locationtech/geomesa/issues/comments{/number}","contents_url":"https://api.github.com/repos/locationtech/geomesa/contents/{+path}","compare_url":"https://api.github.com/repos/locationtech/geomesa/compare/{base}...{head}","merges_url":"https://api.github.com/repos/locationtech/geomesa/merges","archive_url":"https://api.github.com/repos/locationtech/geomesa/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/locationtech/geomesa/downloads","issues_url":"https://api.github.com/repos/locationtech/geomesa/issues{/number}","pulls_url":"https://api.github.com/repos/locationtech/geomesa/pulls{/number}","milestones_url":"https://api.github.com/repos/locationtech/geomesa/milestones{/number}","notifications_url":"https://api.github.com/repos/locationtech/geomesa/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/locationtech/geomesa/labels{/name}","releases_url":"https://api.github.com/repos/locationtech/geomesa/releases{/id}","created_at":"2013-09-30T20:13:46Z","updated_at":"2015-06-02T09:00:18Z","pushed_at":"2015-06-02T20:12:24Z","git_url":"git://github.com/locationtech/geomesa.git","ssh_url":"git@github.com:locationtech/geomesa.git","clone_url":"https://github.com/locationtech/geomesa.git","svn_url":"https://github.com/locationtech/geomesa","homepage":null,"size":22800,"stargazers_count":77,"watchers_count":77,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":46,"mirror_url":null,"open_issues_count":12,"forks":46,"open_issues":12,"watchers":77,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/locationtech/geomesa/pulls/596"},"html":{"href":"https://github.com/locationtech/geomesa/pull/596"},"issue":{"href":"https://api.github.com/repos/locationtech/geomesa/issues/596"},"comments":{"href":"https://api.github.com/repos/locationtech/geomesa/issues/596/comments"},"review_comments":{"href":"https://api.github.com/repos/locationtech/geomesa/pulls/596/comments"},"review_comment":{"href":"https://api.github.com/repos/locationtech/geomesa/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/locationtech/geomesa/pulls/596/commits"},"statuses":{"href":"https://api.github.com/repos/locationtech/geomesa/statuses/5a866421a301c2cfe480376f240049333050d44b"}}}}