Skip to content
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

Jump to bottom

fix reactor upgrade issue with bulk #20083

Merged
merged 2 commits into from
Mar 24, 2021
Merged

fix reactor upgrade issue with bulk #20083

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a74f70c
fix reactor upgrade issue with bulk
moderakh Mar 24, 2021
02d4056
NIT: Removing unnecessary import
FabianMeiswinkel Mar 24, 2021
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
11 changes: 5 additions & 6 deletions ...cosmos/azure-cosmos-spark_3-1_2-12/src/main/scala/com/azure/cosmos/spark/BulkWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import com.azure.cosmos.spark.BulkWriter.MaxNumberOfThreadsPerCPUCore
import com.azure.cosmos.{BulkOperations, CosmosAsyncContainer, CosmosBulkOperationResponse, CosmosException, CosmosItemOperation}
import com.fasterxml.jackson.databind.node.ObjectNode
import reactor.core.Disposable
import reactor.core.publisher.EmitterProcessor
import reactor.core.publisher.Sinks
import reactor.core.scala.publisher.SMono.PimpJFlux
import reactor.core.scala.publisher.{SFlux, SMono}
import reactor.core.scheduler.Schedulers
Expand Down Expand Up @@ -38,8 +38,7 @@ class BulkWriter(container: CosmosAsyncContainer,
private val pendingTasksCompleted = lock.newCondition
private val activeTasks = new AtomicInteger(0)
private val errorCaptureFirstException = new AtomicReference[Throwable]()

private val bulkInputEmitter: EmitterProcessor[CosmosItemOperation] = EmitterProcessor.create[CosmosItemOperation]()
private val bulkInputEmitter: Sinks.Many[CosmosItemOperation] = Sinks.many().unicast().onBackpressureBuffer()
// TODO: moderakh discuss the context issue in the core SDK bulk api with the team.
// public <TContext> Flux<CosmosBulkOperationResponse<TContext>> processBulkOperations(
// Flux<CosmosItemOperation> operations,
Expand All @@ -60,7 +59,7 @@ class BulkWriter(container: CosmosAsyncContainer,

private val subscriptionDisposable: Disposable = {
val bulkOperationResponseFlux: SFlux[CosmosBulkOperationResponse[Object]] =
container.processBulkOperations[Object](bulkInputEmitter).asScala
container.processBulkOperations[Object](bulkInputEmitter.asFlux()).asScala

bulkOperationResponseFlux.subscribe(
resp => {
Expand Down Expand Up @@ -173,7 +172,7 @@ class BulkWriter(container: CosmosAsyncContainer,
}

activeOperations.put(bulkItemOperation, operationContext)
bulkInputEmitter.onNext(bulkItemOperation)
bulkInputEmitter.tryEmitNext(bulkItemOperation)
}

// the caller has to ensure that after invoking this method scheduleWrite doesn't get invoked
Expand Down Expand Up @@ -204,7 +203,7 @@ class BulkWriter(container: CosmosAsyncContainer,

logInfo("invoking bulkInputEmitter.onComplete()")
semaphore.release(activeTasks.get())
bulkInputEmitter.onComplete()
bulkInputEmitter.tryEmitComplete()

// which error to report?
if (errorCaptureFirstException.get() != null) {
Expand Down
4 changes: 2 additions & 2 deletions ...s/azure-cosmos-spark_3-1_2-12/src/test/scala/com/azure/cosmos/spark/BulkWriterITest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ class BulkWriterITest extends IntegrationSpec with CosmosClient with AutoCleanab
val bulkWriter = new BulkWriter(container, writeConfig)

val items = mutable.Map[String, ObjectNode]()
for(_ <- 0 until 500) {
for(_ <- 0 until 5000) {
val item = getItem(UUID.randomUUID().toString)
val id = item.get("id").textValue()
items += (id -> item)
Expand All @@ -55,7 +55,7 @@ class BulkWriterITest extends IntegrationSpec with CosmosClient with AutoCleanab
val bulkWriter = new BulkWriter(container, writeConfig)
val items = new mutable.HashMap[String, mutable.Set[ObjectNode]] with mutable.MultiMap[String, ObjectNode]

for(i <- 0 until 500) {
for(i <- 0 until 5000) {
val item = getItem((i % 100).toString)
val id = item.get("id").textValue()
items.addBinding(id, item)
Expand Down