Skip to content

[SPARK-18682][SS] Batch Source for Kafka #16686

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ghost wants to merge 23 commits into from
Closed

[SPARK-18682][SS] Batch Source for Kafka #16686

Show file tree
Hide file tree
Changes from 16 commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
d371758
add kafka relation and refactor kafka source
Jan 20, 2017
b6c3055
update
Jan 20, 2017
4c81812
update
Jan 20, 2017
ab02a4c
single kafka provider for both stream and batch
Jan 20, 2017
e6b57ed
added uninterruptible thread version of kafka offset reader
Jan 24, 2017
ff94ed8
added uninterruptible thread version of kafka offset reader
Jan 24, 2017
f8fd34c
update tests
Jan 24, 2017
41271e2
resolve conflicts in KafakSource
Jan 24, 2017
74d96fc
update comments
Jan 24, 2017
d31fc81
address comments from @zsxwing
Jan 25, 2017
1db1649
update
Jan 25, 2017
3b0d48b
Merge branch 'master' of https://github.com/apache/spark into SPARK-1…
Jan 25, 2017
a5b0269
address comments from @zsxwing
Jan 26, 2017
c08c01f
late binding offsets
Jan 27, 2017
79d335e
update to late binding logic
Jan 27, 2017
b597cf1
remove kafka log4j debug
Jan 27, 2017
2487a72
address comments from @zsxwing
Jan 31, 2017
789d3af
update
Jan 31, 2017
5b48fc6
address comments from @tdas
Feb 3, 2017
5776009
address feedback from @tdas and @sxwing
Feb 3, 2017
aef89bc
fix indent
Feb 4, 2017
4e56f8c
address comments from @zsxwing
Feb 7, 2017
3bc7c4c
address comments from @zsxwing
Feb 7, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 13 additions & 12 deletions ...nal/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/CachedKafkaConsumer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ private[kafka010] case class CachedKafkaConsumer private(

private val groupId = kafkaParams.get(ConsumerConfig.GROUP_ID_CONFIG).asInstanceOf[String]

private var consumer = createConsumer
var rawConsumer = createConsumer
Copy link
Contributor

@tdas tdas Feb 2, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

exposing internal var is generally not a good idea. A better approach is be to add the necessary methods (for which you need the consumer) in the class CachedKafkaConsumer.

Copy link
Contributor

@tdas tdas Feb 2, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

and why renamed?


/** Iterator to the already fetch data */
private var fetchedData = ju.Collections.emptyIterator[ConsumerRecord[Array[Byte], Array[Byte]]]
Expand Down Expand Up @@ -224,8 +224,8 @@ private[kafka010] case class CachedKafkaConsumer private(

/** Create a new consumer and reset cached states */
private def resetConsumer(): Unit = {
consumer.close()
consumer = createConsumer
rawConsumer.close()
rawConsumer = createConsumer
resetFetchedData()
}

Expand Down Expand Up @@ -271,15 +271,15 @@ private[kafka010] case class CachedKafkaConsumer private(
}
}

private def close(): Unit = consumer.close()
private def close(): Unit = rawConsumer.close()

private def seek(offset: Long): Unit = {
logDebug(s"Seeking to $groupId $topicPartition $offset")
consumer.seek(topicPartition, offset)
rawConsumer.seek(topicPartition, offset)
}

private def poll(pollTimeoutMs: Long): Unit = {
val p = consumer.poll(pollTimeoutMs)
val p = rawConsumer.poll(pollTimeoutMs)
val r = p.records(topicPartition)
logDebug(s"Polled $groupId ${p.partitions()} ${r.size}")
fetchedData = r.iterator
Expand All @@ -290,10 +290,10 @@ private[kafka010] case class CachedKafkaConsumer private(
* and the latest offset.
*/
private def getAvailableOffsetRange(): (Long, Long) = {
consumer.seekToBeginning(Set(topicPartition).asJava)
val earliestOffset = consumer.position(topicPartition)
consumer.seekToEnd(Set(topicPartition).asJava)
val latestOffset = consumer.position(topicPartition)
rawConsumer.seekToBeginning(Set(topicPartition).asJava)
val earliestOffset = rawConsumer.position(topicPartition)
rawConsumer.seekToEnd(Set(topicPartition).asJava)
val latestOffset = rawConsumer.position(topicPartition)
(earliestOffset, latestOffset)
}
}
Expand Down Expand Up @@ -334,14 +334,15 @@ private[kafka010] object CachedKafkaConsumer extends Logging {
def getOrCreate(
topic: String,
partition: Int,
kafkaParams: ju.Map[String, Object]): CachedKafkaConsumer = synchronized {
kafkaParams: ju.Map[String, Object],
reuse: Boolean): CachedKafkaConsumer = synchronized {
Copy link
Contributor

@tdas tdas Feb 2, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this mean reuse existing one, OR allow reuse in future?

Copy link
Author

@ghost ghost Feb 2, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reuse existing. I changed the name to reuseExistingIfPresent.

val groupId = kafkaParams.get(ConsumerConfig.GROUP_ID_CONFIG).asInstanceOf[String]
val topicPartition = new TopicPartition(topic, partition)
val key = CacheKey(groupId, topicPartition)

// If this is reattempt at running the task, then invalidate cache and start with
// a new consumer
if (TaskContext.get != null && TaskContext.get.attemptNumber > 1) {
if (!reuse || TaskContext.get != null && TaskContext.get.attemptNumber > 1) {
val removedConsumer = cache.remove(key)
if (removedConsumer != null) {
removedConsumer.close()
Expand Down
Loading