Persistence of async indexing requests

[heinrichcohn]

Hello

I've been using conductor for a few months and i have noticed a few things that should be improved in the async indexing area.

The current implementation of the async indexing is a little bit problematic because:

1. Queueing of indexing requests is done in memory so in case of a failure all these requests will be lost, causing inconsistency between the data stored in the ExecutionDB and in ES.

2. In case of high throughput of indexing requests, the async indexing queue will be filled very quickly and any new request will be dropped, causing again inconsistency between the data stored in the ExecutionDB and in ES.

New approch
I've been working on a new approach which consists on a new component called "AsyncIndexer" that can be injected into the indexDao and it will handle the async indexing.

This AsyncIndexer will have two implementations:

1. The default implementation. Current existing logic using completable-futures + threadpool.
2. Implementation that uses RedisStreams (it requires upgrade of jedis)

The idea of implementation (2) is as follows:

Implementation (1) will be added to conductor-core project
Implementation (2) will be added to redis-persistence project

@apanicker-nflx I would like to know what you think about this approach

Your feedbacks are welcomed

[apanicker-nflx]

@heinrichcohn Your observations are spot-on and the proposal of externalizing the queues into a durable queue implementation is most definitely the right choice.
Wrt the implementation, I would suggest keeping the options open to be configurable (as with most-things in Conductor). Similar to using RedisStreams, we could use a Kafka topic as the intermediate shock-absorbing queue layer (Internally, we are using Kafka for this purpose). It seems that you are alluding to a similar approach.

Do you have an interface blueprint in mind? Please share the same and we can work out the details.

[heinrichcohn]

Hi

This is the interface i have implemented:

public interface AsyncIndexer {
void indexWorkflow(Workflow workflow);
void indexTask(Task task);
void addEventExecution(EventExecution eventExecution);
void addMessage(String queue, Message message);
void addTaskExecutionLogs(List logs);
void removeWorkflow(String workflowId);
}

I'm using the same method names as in the synchronous methods in IndexDao.

In each method a stream message (Map object) will be created and added to redis streams

1. Example (indexWorkflow):
   {
   "op" : "INDEX",
   "workflow" : { [workflow object] }
   }

2. Example (removeWorkflow):
   {
   "op" : "REMOVE",
   "workflowId" : "[workflowId]"
   }

Streams that will be created:

• workflows_stream
• task_stream
• event_exec_stream
• message_stream
• task_exec_logs_stream

Regarding error handling, i won't change the current logic in ES index Dao that just writes to the logs when an indexing operation failed. This is the current login in the sync api and will be also in the async api. In a future PR we can modify the error handling logic, for both sync/async, and in a case of a failure a message can be requeued.

Thanks

[apanicker-nflx]

@heinrichcohn Thanks for detailing out the approach. Upon further deliberation, we think a cleaner approach would be to break up the current IndexDAO interface into 2 separate ones: IndexDAO and SearchDAO.

The SearchDAO as the name suggests would handle all the read queries against the ES index.

The IndexDAO would handle all the write requests to the index. This also allows us to get rid of the extensive checks in the ExecutionDAOFacade for isAsyncIndexingEnabled.
The indexDAO itself can be implemented as a synchronous write to ES/ asynchronous write to ES using in-memory queues/ asynchronous indexer using RedisStreams(as proposed)/ any other implementation. The segregation of the read and write operations into separate interfaces here would help to define/provide cleaner contracts and configuration-driven implementation choices.

Please let me know your thoughts on this approach.