Ривер и клиент

111 · August 12, 2015, 11:27am

Добрый день!

Возникла проблема с ривером следующего рода: при отправке запроса типа PutMappingRequest (при существующем индексе) ривер виснет.

Пример кода:

public class ObjectsRiver extends AbstractRiverComponent implements River {

@Inject
public ObjectsRiver(RiverName riverName, RiverSettings settings,
		Client client, ThreadPool threadPool) {
	super(riverName, settings);
	client.admin()
			.indices()
			.putMapping(new PutMappingRequest("articles")
						.type("article")
						.source("{\"article\":{\"properties\":{\"id\":{\"type\":\"string\",\"index\":\"not_analyzed\"},\"text\":{\"type\":\"string\"}}}}"))
			.actionGet();
}
...
}

Дополнительные замечания:

В логах ничего нету.
При принудительной перезагрузке в логах может появиться сообщение об InterruptedException и о том, что программа не дождалась завершения операции.
Тестировалось на 1.4.3 и на 1.6.2.
Подобная проблема возникает и при запросах типа IndicesExistsRequest.

Igor_Motov · August 12, 2015, 2:54pm

А стек печатается с InterruptedException? Если да, то пришлите его сюда с полным сообщением об ошибке.

111 · August 19, 2015, 12:03pm

Присылаю лог, начиная с момента запуска ривера и заканчивая принудительной остановкой ноды (замечу, что команда типа "POST /_cluster/nodes/_shutdown" не останавливает все потоки при зависнувшем ривере):

[2015-08-19 14:49:49,718][INFO ][cluster.metadata ] [es1] [_river] creating index, cause [auto(index api)], templates , s hards [1]/[1], mappings [river1]
[2015-08-19 14:49:49,973][INFO ][cluster.metadata ] [es1] [_river] update_mapping [river1] (dynamic)
[2015-08-19 14:49:50,909][INFO ][river ] [es1] rivers have been deprecated. Read https://www.elastic.co/blog/dep recating_rivers
[2015-08-19 14:52:09,152][INFO ][action.admin.cluster.node.shutdown] [es1] [cluster_shutdown]: requested, shutting down in [1s]
[2015-08-19 14:52:10,154][INFO ][action.admin.cluster.node.shutdown] [es1] [cluster_shutdown]: done shutting down all nodes except master, proceeding to master
[2015-08-19 14:52:10,155][INFO ][action.admin.cluster.node.shutdown] [es1] shutting down in [200ms]
[2015-08-19 14:52:10,357][INFO ][action.admin.cluster.node.shutdown] [es1] initiating requested shutdown...
[2015-08-19 14:52:10,357][INFO ][node ] [es1] stopping ...
[2015-08-19 14:52:10,454][INFO ][node ] [es1] stopped
[2015-08-19 14:52:10,454][INFO ][node ] [es1] closing ...
[2015-08-19 14:52:20,459][INFO ][node ] [es1] closed

111 · August 19, 2015, 12:03pm

[2015-08-19 14:52:20,461][WARN ][river ] [es1] failed to create river [objects_river][river1]
org.elasticsearch.common.inject.CreationException: Guice creation errors:

Error injecting constructor, org.elasticsearch.ElasticsearchIllegalStateException: Future got interrupted
at by.ias.es.rivers.ObjectsRiver.(Unknown Source)
while locating by.ias.es.rivers.ObjectsRiver
while locating org.elasticsearch.river.River

1 error
at org.elasticsearch.common.inject.internal.Errors.throwCreationExceptionIfErrorsExist(Errors.java:344)
at org.elasticsearch.common.inject.InjectorBuilder.injectDynamically(InjectorBuilder.java:178)
at org.elasticsearch.common.inject.InjectorBuilder.build(InjectorBuilder.java:110)
at org.elasticsearch.common.inject.InjectorImpl.createChildInjector(InjectorImpl.java:131)
at org.elasticsearch.common.inject.ModulesBuilder.createChildInjector(ModulesBuilder.java:69)
at org.elasticsearch.river.RiversService.createRiver(RiversService.java:141)
at org.elasticsearch.river.RiversService$ApplyRivers$2.onResponse(RiversService.java:274)
at org.elasticsearch.river.RiversService$ApplyRivers$2.onResponse(RiversService.java:268)
at org.elasticsearch.action.support.TransportAction$ThreadedActionListener$1.run(TransportAction.java:113)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.elasticsearch.ElasticsearchIllegalStateException: Future got interrupted
at org.elasticsearch.action.support.AdapterActionFuture.actionGet(AdapterActionFuture.java:47)
at by.ias.es.rivers.ObjectsRiver.(ObjectsRiver.java:32)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
at org.elasticsearch.common.inject.DefaultConstructionProxyFactory$1.newInstance(DefaultConstructionProxyFactory.java:54)
at org.elasticsearch.common.inject.ConstructorInjector.construct(ConstructorInjector.java:86)
at org.elasticsearch.common.inject.ConstructorBindingImpl$Factory.get(ConstructorBindingImpl.java:98)
at org.elasticsearch.common.inject.FactoryProxy.get(FactoryProxy.java:52)
at org.elasticsearch.common.inject.ProviderToInternalFactoryAdapter$1.call(ProviderToInternalFactoryAdapter.java:45)
at org.elasticsearch.common.inject.InjectorImpl.callInContext(InjectorImpl.java:837)
at org.elasticsearch.common.inject.ProviderToInternalFactoryAdapter.get(ProviderToInternalFactoryAdapter.java:42)
at org.elasticsearch.common.inject.Scopes$1$1.get(Scopes.java:57)
at org.elasticsearch.common.inject.InternalFactoryToProviderAdapter.get(InternalFactoryToProviderAdapter.java:45)
at org.elasticsearch.common.inject.InjectorBuilder$1.call(InjectorBuilder.java:200)
at org.elasticsearch.common.inject.InjectorBuilder$1.call(InjectorBuilder.java:193)
at org.elasticsearch.common.inject.InjectorImpl.callInContext(InjectorImpl.java:830)
at org.elasticsearch.common.inject.InjectorBuilder.loadEagerSingletons(InjectorBuilder.java:193)
at org.elasticsearch.common.inject.InjectorBuilder.injectDynamically(InjectorBuilder.java:175)
... 10 more
Caused by: java.lang.InterruptedException
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:996)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1303)
at org.elasticsearch.common.util.concurrent.BaseFuture$Sync.get(BaseFuture.java:278)
at org.elasticsearch.common.util.concurrent.BaseFuture.get(BaseFuture.java:117)
at org.elasticsearch.action.support.AdapterActionFuture.actionGet(AdapterActionFuture.java:45)
... 29 more

111 · August 19, 2015, 12:05pm

[2015-08-19 14:52:20,465][WARN ][river ] [es1] failed to write failed status for river creation
org.elasticsearch.common.inject.CreationException: Guice creation errors:

Error injecting constructor, org.elasticsearch.ElasticsearchIllegalStateException: Future got interrupted
at by.ias.es.rivers.ObjectsRiver.(Unknown Source)
while locating by.ias.es.rivers.ObjectsRiver
while locating org.elasticsearch.river.River

1 error
at org.elasticsearch.common.inject.internal.Errors.throwCreationExceptionIfErrorsExist(Errors.java:344)
at org.elasticsearch.common.inject.InjectorBuilder.injectDynamically(InjectorBuilder.java:178)
at org.elasticsearch.common.inject.InjectorBuilder.build(InjectorBuilder.java:110)
at org.elasticsearch.common.inject.InjectorImpl.createChildInjector(InjectorImpl.java:131)
at org.elasticsearch.common.inject.ModulesBuilder.createChildInjector(ModulesBuilder.java:69)
at org.elasticsearch.river.RiversService.createRiver(RiversService.java:141)
at org.elasticsearch.river.RiversService$ApplyRivers$2.onResponse(RiversService.java:274)
at org.elasticsearch.river.RiversService$ApplyRivers$2.onResponse(RiversService.java:268)
at org.elasticsearch.action.support.TransportAction$ThreadedActionListener$1.run(TransportAction.java:113)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.elasticsearch.ElasticsearchIllegalStateException: Future got interrupted
at org.elasticsearch.action.support.AdapterActionFuture.actionGet(AdapterActionFuture.java:47)
at by.ias.es.rivers.ObjectsRiver.(ObjectsRiver.java:32)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
at org.elasticsearch.common.inject.DefaultConstructionProxyFactory$1.newInstance(DefaultConstructionProxyFactory.java:54)
at org.elasticsearch.common.inject.ConstructorInjector.construct(ConstructorInjector.java:86)
at org.elasticsearch.common.inject.ConstructorBindingImpl$Factory.get(ConstructorBindingImpl.java:98)
at org.elasticsearch.common.inject.FactoryProxy.get(FactoryProxy.java:52)
at org.elasticsearch.common.inject.ProviderToInternalFactoryAdapter$1.call(ProviderToInternalFactoryAdapter.java:45)
at org.elasticsearch.common.inject.InjectorImpl.callInContext(InjectorImpl.java:837)
at org.elasticsearch.common.inject.ProviderToInternalFactoryAdapter.get(ProviderToInternalFactoryAdapter.java:42)
at org.elasticsearch.common.inject.Scopes$1$1.get(Scopes.java:57)
at org.elasticsearch.common.inject.InternalFactoryToProviderAdapter.get(InternalFactoryToProviderAdapter.java:45)
at org.elasticsearch.common.inject.InjectorBuilder$1.call(InjectorBuilder.java:200)
at org.elasticsearch.common.inject.InjectorBuilder$1.call(InjectorBuilder.java:193)
at org.elasticsearch.common.inject.InjectorImpl.callInContext(InjectorImpl.java:830)
at org.elasticsearch.common.inject.InjectorBuilder.loadEagerSingletons(InjectorBuilder.java:193)
at org.elasticsearch.common.inject.InjectorBuilder.injectDynamically(InjectorBuilder.java:175)
... 10 more
Caused by: java.lang.InterruptedException
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:996)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1303)
at org.elasticsearch.common.util.concurrent.BaseFuture$Sync.get(BaseFuture.java:278)
at org.elasticsearch.common.util.concurrent.BaseFuture.get(BaseFuture.java:117)
at org.elasticsearch.action.support.AdapterActionFuture.actionGet(AdapterActionFuture.java:45)
... 29 more

(В нескольких сообщениях из-за ограничений на размер)

Igor_Motov · August 20, 2015, 8:24pm

А вы не пробовали эту операцию из конструктора в start() перенести? Так же виснет? Вообще в 2.0 ривер убрали. Так что если вы новое приложение разрабатываете, то я бы не рекомендовал с ними связываться.

111 · August 21, 2015, 7:03am

Пробовал, не помогло. Спасибо за предложение.