Cannot flush non-initialized write operation

[supernova-start]

What kind an issue is this?

• [ *] Bug report. If you’ve found a bug, please provide a code snippet or test to reproduce it below.
  The easier it is to track down the bug, the faster it is solved.
• Feature Request. Start by telling us what problem you’re trying to solve.
  Often a solution already exists! Don’t send pull requests to implement new features without
  first getting our support. Sometimes we leave features out on purpose to keep the project small.

Issue description

Description

When and strom integration, if not data in some time ,the storm supervisor process will crash

Steps to reproduce

Code:

Test/code snippet

Strack trace:
org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: Cannot flush non-initialized write operation
at org.elasticsearch.hadoop.util.Assert.isTrue(Assert.java:60) ~[mystormdemo-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?]
at org.elasticsearch.hadoop.rest.RestRepository.flush(RestRepository.java:202) ~[mystormdemo-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?]
at org.elasticsearch.storm.EsBolt.flushNoAck(EsBolt.java:193) ~[mystormdemo-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?]
at org.elasticsearch.storm.EsBolt.flush(EsBolt.java:155) ~[mystormdemo-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?]
at org.elasticsearch.storm.EsBolt.cleanup(EsBolt.java:200) ~[mystormdemo-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?]
at org.apache.storm.executor.ExecutorShutdown.shutdown(ExecutorShutdown.java:121) ~[storm-client-2.0.0.jar:2.0.0]
at org.apache.storm.daemon.worker.Worker.shutdown(Worker.java:456) ~[storm-client-2.0.0.jar:2.0.0]
at org.apache.storm.ProcessSimulator.killProcess(ProcessSimulator.java:62) ~[storm-server-2.0.0.jar:2.0.0]
at org.apache.storm.daemon.supervisor.LocalContainer.kill(LocalContainer.java:66) ~[storm-server-2.0.0.jar:2.0.0]
at org.apache.storm.daemon.supervisor.Slot.killContainerFor(Slot.java:269) ~[storm-server-2.0.0.jar:2.0.0]
at org.apache.storm.daemon.supervisor.Slot.handleRunning(Slot.java:724) ~[storm-server-2.0.0.jar:2.0.0]
at org.apache.storm.daemon.supervisor.Slot.stateMachineStep(Slot.java:218) ~[storm-server-2.0.0.jar:2.0.0]
at org.apache.storm.daemon.supervisor.Slot.run(Slot.java:931) [storm-server-2.0.0.jar:2.0.0]

Stack trace goes here

Version Info

OS: : Ubuntu
JVM : 12.0.2
Hadoop/Spark:
ES-Hadoop :
ES :

Feature description

After check the code ,we find the fucntion lazyInitWriting is the reason in org.elasticsearch.hadoop.rest.RestRepository class , it don't have be called if the storm don't have data flow。

[DaveyDevOps]

We saw a similar issue working with Storm 1.1.0 and Elastic 7.3

Thanks to @kiranyagna for the explanation.

In our case

It happens when you have tick tuples generated by storm before a document to be indexed, this makes it to call flush when tick tuple arrives without calling writeToIndex and it checks on

elasticsearch-hadoop/mr/src/main/java/org/elasticsearch/hadoop/rest/RestRepository.java

Line 197 in 6c82d4b

Assert.isTrue(writeInitialized, "Cannot flush non-initialized write operation");

which fails

I got to the root cause of this problem which is a setting in yaml file es.storm.bolt.tick.tuple.flush: true
disabling tick tuple and the flush setting in yaml file successfully started the topology and didn't throw this exception.

[DaveyDevOps]

There were changes to public BulkResponse tryFlush() from
Add "Dead Letter Handlers" for bulk write failures #1095

Currently the problem we are facing is, when the tick tuple arrives before any document for indexing, it tries to flush. But writeInitialized flag is not set to true until document arrives which results in throwing "Cannot flush non-initialized write operation" exception. We can’t get away with not generating tick tuples as it is needed to flush ES docs frequently instead of waiting for es.storm.bolt.flush.entries.size to kick in.

[supernova-start]

I think the Assert function is not appropriate, this is my changes in org.elasticsearch.hadoop.rest.RestRepository file。It work better now。
public BulkResponse tryFlush() {
- Assert.isTrue(writeInitialized, "Cannot flush non-initialized write operation");
+ if (!writeInitialized) {
+ return BulkResponse.complete();
+ }
return bulkProcessor.tryFlush();
}

public void flush() {
    **- Assert.isTrue(writeInitialized, "Cannot flush non-initialized write operation");
    + if (!writeInitialized) {
    +       return;
    + }**    
     bulkProcessor.flush();
}

This changes is correct？

[yeongchuin]

Hi, any update on this issue?

[kiranyagna]

I believe this problem still exists. I had to put a check for tick tuples to bypass this issue. I am not sure if anyone fixed it.

…

On Wed, Jul 8, 2020 at 10:48 PM yeongchuin ***@***.***> wrote: Hi, any update on this issue? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1357 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AJTJLJXO5GN2K4EEBS25C73R2U4Z3ANCNFSM4IYWBFSA> .

-- Thanks & Regards, Kiran Yagnavajhala +1-669-220-8654