Add cached receiver #303
Add cached receiver #303
Conversation
There was a problem hiding this comment.
Am exited to see cachedReceiver going live! You can run Spark jobs in full speed! Please fix these few changes.. I wanted to just click on approve - but, I also know @sjkwak - will merge if I do :).
| val prefetchCounts = new FixedList[Int](5) | ||
|
|
||
| private var _client: EventHubClient = _ | ||
| private def client: EventHubClient = { |
There was a problem hiding this comment.
Why? There should only be one thread in this path, right?
| val consumerGroup = ehConf.consumerGroup.getOrElse(DefaultConsumerGroup) | ||
| val receiverOptions = new ReceiverOptions | ||
| receiverOptions.setReceiverRuntimeMetricEnabled(false) | ||
| receiverOptions.setIdentifier(s"${SparkEnv.get.executorId}-${TaskContext.get.taskAttemptId}") |
There was a problem hiding this comment.
if executorid is like some guid and doesn't indicate - that it is a spark job - consider adding a prefix - hint like - "spark-.." - to indicate that this receiver is coming from spark job.
| _receiver.setPrefetchCount(DefaultPrefetchCount) | ||
| } | ||
|
|
||
| private[this] var _receiver: PartitionReceiver = _ |
There was a problem hiding this comment.
so - CachedEventHubsReceiver cannot be used - unless createReceiver is invoked.
can we use factory pattern here instead ?
| nAndP: NameAndPartition, | ||
| requestSeqNo: SequenceNumber, | ||
| batchSize: Int): EventData = { | ||
| receivers.synchronized { |
There was a problem hiding this comment.
we need something like this - to avoid your getters on the map - conflict with setters - performing a rehash etc.
receivers.sync {
val receiver: PartitionReceiver // <- I used a val here which could be wrong. intent here is to mark it readonly - like a `final` in java - and single assignment
if (!isInit(nAndP)) {
receivers.update(nAndP, CreateEHReceiver())
}
receiver = get(nAndP) // **<- inside sync block**
} receiver.receive(...)| event | ||
| } | ||
|
|
||
| private def logAndUpdatePrefetch(batchSize: Int): Unit = { |
There was a problem hiding this comment.
lets updatePrefetch to latest BatchSize to keep the code simple and behavior predictable.
|
|
||
| type MutableMap[A, B] = scala.collection.mutable.HashMap[A, B] | ||
|
|
||
| private[this] val receivers = new MutableMap[NameAndPartition, CachedEventHubsReceiver]() |
There was a problem hiding this comment.
Key should include namespace
There was a problem hiding this comment.
Will address in #305
| } | ||
|
|
||
| private def get(nAndP: NameAndPartition): CachedEventHubsReceiver = receivers.synchronized { | ||
| logInfo(s"lookup on $nAndP") |
There was a problem hiding this comment.
is this logging for every receive call ?
| nAndP: NameAndPartition, | ||
| requestSeqNo: SequenceNumber, | ||
| batchSize: Int): EventData = { | ||
| receivers.synchronized { |
There was a problem hiding this comment.
nit: feel free to ignore..
one small Idea after our discussion:
val receiver = createIfNotExist(...) // <- this captures all of the isInitialized receiver.update, get into 1 synchronized block ...
receiver.receive()
There was a problem hiding this comment.
override def receive(ehConf: EventHubsConf,
nAndP: NameAndPartition,
requestSeqNo: SequenceNumber,
batchSize: Int): EventData = {
val receiver = createCachedReceiverIfNotExist(ehConf, nAndP, requestSeqNo)
receiver.receive(requestSeqNo, batchSize)
}
def createCachedReceiverIfNotExist(ehConf: EventHubsConf,
nAndP: NameAndPartition,
requestSeqNo: SequenceNumber) : CachedEventHubsReceiver = {
var receiver: CachedEventHubsReceiver = null
receivers.synchronized {
receiver = receivers.getOrElseUpdate(nAndP, new CachedEventHubsReceiver(ehConf, nAndP)) // take care to initialize this lazily
}
receiver
}
This PR adds the cached receiver and cleans up of the client APIs.
The
CachedEventHubsReceiverborrows a connection from theClientConnectionPooland opens a receiver link. This receiver link will stay open until:The average of the last five batch sizes is used as the prefetchCount.
Close #259