Ошибка Couchbase: rx.exceptions.OnErrorThrowable $ OnNextValue: OnError при отправке значения onNext - PullRequest
Новая
       13

Ошибка Couchbase: rx.exceptions.OnErrorThrowable $ OnNextValue: OnError при отправке значения onNext

0 голосов
/ 23 декабря 2018

У меня есть приложение Spring-boot, которое сохраняет данные на Couchbase сервере.

Мне нравится сохранять около 500 тыс. Студентов, использующих этот код: 

for (String student: students) {
  ...
  bucket.upsert(JsonDocument.create(<unique key here>, TTL, JsonObject.fromJson(studentAsJson)));                
}

Это прекрасно работает длянебольшое количество студентов (до 100K + -).

Но всякий раз, когда приложение достигает примерно 110K-140K студентов, я получаю эту ошибку: 

2018-12-23 16:20:57.804 ERROR 17468 --- [nio-8989-exec-6] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is com.couchbase.client.java.error.TemporaryFailureException] with root cause

rx.exceptions.OnErrorThrowable$OnNextValue: OnError while emitting onNext value: com.couchbase.client.core.message.kv.UpsertResponse.class
at rx.exceptions.OnErrorThrowable.addValueAsLastCause(OnErrorThrowable.java:118) ~[rxjava-1.3.8.jar:1.3.8]
at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:73) ~[rxjava-1.3.8.jar:1.3.8]
at rx.observers.Subscribers$5.onNext(Subscribers.java:235) ~[rxjava-1.3.8.jar:1.3.8]
at rx.internal.operators.OnSubscribeDoOnEach$DoOnEachSubscriber.onNext(OnSubscribeDoOnEach.java:101) ~[rxjava-1.3.8.jar:1.3.8]
at rx.internal.producers.SingleProducer.request(SingleProducer.java:65) ~[rxjava-1.3.8.jar:1.3.8]
at rx.internal.producers.ProducerArbiter.setProducer(ProducerArbiter.java:126) ~[rxjava-1.3.8.jar:1.3.8]
at rx.internal.operators.OnSubscribeTimeoutTimedWithFallback$TimeoutMainSubscriber.setProducer(OnSubscribeTimeoutTimedWithFallback.java:155) ~[rxjava-1.3.8.jar:1.3.8]
at rx.Subscriber.setProducer(Subscriber.java:205) ~[rxjava-1.3.8.jar:1.3.8]
at rx.internal.operators.OnSubscribeMap$MapSubscriber.setProducer(OnSubscribeMap.java:102) ~[rxjava-1.3.8.jar:1.3.8]
at rx.Subscriber.setProducer(Subscriber.java:205) ~[rxjava-1.3.8.jar:1.3.8]
at rx.Subscriber.setProducer(Subscriber.java:205) ~[rxjava-1.3.8.jar:1.3.8]
at rx.subjects.AsyncSubject.onCompleted(AsyncSubject.java:103) ~[rxjava-1.3.8.jar:1.3.8]
at com.couchbase.client.core.endpoint.AbstractGenericHandler.completeResponse(AbstractGenericHandler.java:508) ~[core-io-1.7.1.jar:na]
at com.couchbase.client.core.endpoint.AbstractGenericHandler.access$000(AbstractGenericHandler.java:86) ~[core-io-1.7.1.jar:na]
at com.couchbase.client.core.endpoint.AbstractGenericHandler$1.call(AbstractGenericHandler.java:526) ~[core-io-1.7.1.jar:na]
at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) ~[rxjava-1.3.8.jar:1.3.8]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_161]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_161]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) ~[na:1.8.0_161]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) ~[na:1.8.0_161]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_161]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_161]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]

Мой Couchbase работает вDocker на локальном хосте, и мне интересно, если это какая-то реальная проблема или просто ограничение из-за docker & localhost.

Ответы [ 2 ]

0 голосов
/ 02 января 2019

Таким образом, после того, как код достиг «обычной» среды с реальной БД, которая работает не внутри контейнера, проблема исчезла.

0 голосов
/ 24 декабря 2018

Никогда не работал с CouchBase, но похоже, что это проблема с противодавлением (https://github.com/ReactiveX/RxJava/wiki/Backpressure-(2.0)).

). Таким образом, в основном ваш потребитель (в данном случае CouchBase) не может использовать данные так же быстро, как вы его предоставляете.Из того, что видно из журналов, которые вы используете RxJava1. Я бы предложил обновить до RxJava2. Потребуется некоторое время из-за изменений, но у вас будет доступ к Flowable (http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html), который предлагает BackPressure из коробки)..

Установив Flowable с противодавлением, вы сможете позаботиться о проблеме для медленных потребителей, так как это позволит вам подождать, пока потребитель потребит данные, а затем отправит ему новые.

Если вы не можете обновить версию RxJava, стоит взглянуть на эту статью Дэвида Карнока, поскольку в ней показаны способы работы с этими цепями без необходимости обратного давления: https://github.com/ReactiveX/RxJava/wiki/Backpressure

Надеюсь, этопомогает.

...