You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2013/04/18 06:54:12 UTC
[18/27] git commit: kafka-763 delta;
Add an option to replica from the largest offset during unclean
leader election; patched by Swapnil Ghike; reviewed by Jun Rao
kafka-763 delta; Add an option to replica from the largest offset during unclean leader election; patched by Swapnil Ghike; reviewed by Jun Rao
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3b3fb7fe
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3b3fb7fe
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3b3fb7fe
Branch: refs/heads/trunk
Commit: 3b3fb7fed622cc0339c3b9e3fb175e6093e427a3
Parents: 0a92835
Author: Swapnil Ghike <sg...@linkedin.com>
Authored: Wed Mar 13 15:11:15 2013 -0700
Committer: Jun Rao <ju...@gmail.com>
Committed: Wed Mar 13 15:11:15 2013 -0700
----------------------------------------------------------------------
.../kafka/consumer/ConsumerFetcherThread.scala | 7 +------
.../main/scala/kafka/consumer/SimpleConsumer.scala | 1 +
2 files changed, 2 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b3fb7fe/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala b/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
index 1dfc75c..80df1b5 100644
--- a/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
+++ b/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
@@ -58,12 +58,7 @@ class ConsumerFetcherThread(name: String,
case OffsetRequest.LargestTimeString => startTimestamp = OffsetRequest.LatestTime
case _ => startTimestamp = OffsetRequest.LatestTime
}
- val request = OffsetRequest(Map(topicAndPartition -> PartitionOffsetRequestInfo(startTimestamp, 1)))
- val partitionErrorAndOffset = simpleConsumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition)
- val newOffset = partitionErrorAndOffset.error match {
- case ErrorMapping.NoError => partitionErrorAndOffset.offsets.head
- case _ => throw ErrorMapping.exceptionFor(partitionErrorAndOffset.error)
- }
+ val newOffset = simpleConsumer.earliestOrLatestOffset(topicAndPartition, startTimestamp, Request.OrdinaryConsumerId)
val pti = partitionMap(topicAndPartition)
pti.resetFetchOffset(newOffset)
pti.resetConsumeOffset(newOffset)
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b3fb7fe/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/SimpleConsumer.scala b/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
index 5a0784a..1fbdfc3 100644
--- a/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
+++ b/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
@@ -141,6 +141,7 @@ class SimpleConsumer(val host: String,
*/
def earliestOrLatestOffset(topicAndPartition: TopicAndPartition, earliestOrLatest: Long, consumerId: Int): Long = {
val request = OffsetRequest(requestInfo = Map(topicAndPartition -> PartitionOffsetRequestInfo(earliestOrLatest, 1)),
+ clientId = clientId,
replicaId = consumerId)
val partitionErrorAndOffset = getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition)
val offset = partitionErrorAndOffset.error match {