Skip to content

[SPARK-26121] [Structured Streaming] Allow users to define prefix of Kafka's consumer group (group.id) #23103

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
zouzias wants to merge 7 commits into from
Closed

[SPARK-26121] [Structured Streaming] Allow users to define prefix of Kafka's consumer group (group.id) #23103

Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4afd3a4
[SPARK-26121] init commit
Nov 21, 2018
9a09e74
[SPARK-26121] make scalastyle happy
Nov 21, 2018
cac7976
[SPARK-26121] fix javadoc error
Nov 21, 2018
9d9c944
[SPARK-26121] reviewer comments
Nov 21, 2018
39424dd
[SPARK-26121] rename property to groupIdPrefix
Nov 22, 2018
f79ba9f
[SPARK-26121] minor fix
Nov 22, 2018
9196623
[SPARK-26121] [Documentation] init commit
Nov 23, 2018
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
17 changes: 14 additions & 3 deletions ...nal/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ private[kafka010] class KafkaSourceProvider extends DataSourceRegister
// Each running query should use its own group id. Otherwise, the query may be only assigned
// partial data since Kafka will assign partitions to multiple consumers having the same group
// id. Hence, we should generate a unique id for each query.
val uniqueGroupId = s"spark-kafka-source-${UUID.randomUUID}-${metadataPath.hashCode}"
val uniqueGroupId = streamingUniqueGroupId(parameters, metadataPath)

val caseInsensitiveParams = parameters.map { case (k, v) => (k.toLowerCase(Locale.ROOT), v) }
val specifiedKafkaParams =
Expand Down Expand Up @@ -119,7 +119,7 @@ private[kafka010] class KafkaSourceProvider extends DataSourceRegister
// Each running query should use its own group id. Otherwise, the query may be only assigned
// partial data since Kafka will assign partitions to multiple consumers having the same group
// id. Hence, we should generate a unique id for each query.
val uniqueGroupId = s"spark-kafka-source-${UUID.randomUUID}-${metadataPath.hashCode}"
val uniqueGroupId = streamingUniqueGroupId(parameters, metadataPath)

val caseInsensitiveParams = parameters.map { case (k, v) => (k.toLowerCase(Locale.ROOT), v) }
val specifiedKafkaParams =
Expand Down Expand Up @@ -159,7 +159,7 @@ private[kafka010] class KafkaSourceProvider extends DataSourceRegister
// Each running query should use its own group id. Otherwise, the query may be only assigned
// partial data since Kafka will assign partitions to multiple consumers having the same group
// id. Hence, we should generate a unique id for each query.
val uniqueGroupId = s"spark-kafka-source-${UUID.randomUUID}-${metadataPath.hashCode}"
val uniqueGroupId = streamingUniqueGroupId(parameters, metadataPath)

val caseInsensitiveParams = parameters.map { case (k, v) => (k.toLowerCase(Locale.ROOT), v) }
val specifiedKafkaParams =
Expand Down Expand Up @@ -538,6 +538,17 @@ private[kafka010] object KafkaSourceProvider extends Logging {
.setIfUnset(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 65536: java.lang.Integer)
.build()

/**
* Returns a unique consumer group (group.id), allowing the user to set the prefix of
* the consumer group
*/
private def streamingUniqueGroupId(parameters: Map[String, String],
metadataPath: String): String = {
val groupIdPrefix = parameters
.getOrElse("group.id.prefix", "spark-kafka-source")
s"${groupIdPrefix}-${UUID.randomUUID}-${metadataPath.hashCode}"
}

/** Class to conveniently update Kafka config params, while logging the changes */
private case class ConfigUpdater(module: String, kafkaParams: Map[String, String]) {
private val map = new ju.HashMap[String, Object](kafkaParams.asJava)
Expand Down