Service layer and Transport Handlers #8

eirsep · 2020-11-23T12:00:01Z

Issue #3

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

…x in partial result holder

…inary

Bukhtawar

Thanks Surya few comments from the first pass, will take a detailed look

Bukhtawar · 2020-11-30T16:34:09Z

...endistroforelasticsearch/search/async/context/persistence/AsyncSearchPersistenceContext.java

+        BytesReference bytesReference =
+                BytesReference.fromByteBuffer(ByteBuffer.wrap(Base64.getUrlDecoder().decode(asyncSearchPersistenceModel.getResponse())));
+        try {
+            NamedWriteableAwareStreamInput wrapperStreamInput = new NamedWriteableAwareStreamInput(bytesReference.streamInput(),
+                    namedWriteableRegistry);
+            SearchResponse asyncSearchResponse = new SearchResponse(wrapperStreamInput);
+            wrapperStreamInput.close();
+            return asyncSearchResponse;
+        } catch (IOException e) {
+            logger.error("Failed to parse search response " + asyncSearchPersistenceModel.getResponse(), e);
+            return null;


Use try-with-resource to avoid resource leaks

Bukhtawar · 2020-11-30T16:34:33Z

...endistroforelasticsearch/search/async/context/persistence/AsyncSearchPersistenceContext.java

+        try {
+            NamedWriteableAwareStreamInput wrapperStreamInput = new NamedWriteableAwareStreamInput(bytesReference.streamInput(),
+                    namedWriteableRegistry);
+            ElasticsearchException exception =wrapperStreamInput.readException();
+            wrapperStreamInput.close();
+            return exception;
+        } catch (IOException e) {
+            logger.error("Failed to parse search error " + asyncSearchPersistenceModel.getError(), e);
+            return null;


Try with resource here as well

Bukhtawar · 2020-11-30T16:36:26Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/plugin/AsyncSearchPlugin.java

-                (s, e) -> {}, (contextId, listener) -> listener.onContextPersisted(contextId), SearchResponsePersistedEvent.class));
+                (s, e) -> {
+                }, (contextId, listener) -> listener.onContextPersisted(contextId), SearchResponsePersistedEvent.class));

        stateMachine.registerTransition(new AsyncSearchTransition<>(FAILED, PERSISTED,
-                (s, e) -> {}, (contextId, listener) -> listener.onContextPersisted(contextId), SearchResponsePersistedEvent.class));
+                (s, e) -> {
+                }, (contextId, listener) -> listener.onContextPersisted(contextId), SearchResponsePersistedEvent.class));

        //persist failed
        stateMachine.registerTransition(new AsyncSearchTransition<>(SUCCEEDED, PERSIST_FAILED,
-                (s, e) -> {}, (contextId, listener) -> listener.onContextPersistFailed(contextId), SearchResponsePersistFailedEvent.class));
+                (s, e) -> {
+                }, (contextId, listener) -> listener.onContextPersistFailed(contextId), SearchResponsePersistFailedEvent.class));

        stateMachine.registerTransition(new AsyncSearchTransition<>(FAILED, PERSIST_FAILED,
-                (s, e) -> {}, (contextId, listener) -> listener.onContextPersistFailed(contextId), SearchResponsePersistFailedEvent.class));
+                (s, e) -> {
+                }, (contextId, listener) -> listener.onContextPersistFailed(contextId), SearchResponsePersistFailedEvent.class));


Lets addd a TODO to populate respective EVents here

Bukhtawar · 2020-11-30T16:37:39Z

...a/com/amazon/opendistroforelasticsearch/search/async/processor/AsyncSearchPostProcessor.java

+                    if (asyncSearchContext.getAsyncSearchState() == AsyncSearchState.DELETED) {
+                        logger.debug("Async search context {} has been moved to DELETED while waiting to acquire permits for post " +
+                                "processing", asyncSearchContext.getAsyncSearchId());
+                        return;
+                    }


Lets release a permit here

Bukhtawar · 2020-11-30T16:38:27Z

...a/com/amazon/opendistroforelasticsearch/search/async/processor/AsyncSearchPostProcessor.java

+                        return;
+                    }
+                    asyncSearchPersistenceService.storeResponse(asyncSearchContext.getAsyncSearchId(),
+                            persistenceModel, ActionListener.wrap(


ActionListener.runafter(ActionListener.wrap(..., releasable::close) instead

Bukhtawar · 2020-11-30T16:41:44Z

...mazon/opendistroforelasticsearch/search/async/transport/TransportAsyncSearchFetchAction.java

+        protected void doRun() {
+            //should we look at the task status and retry on forwarding the request if the search is still RUNNING based
+            //on task status
+            ClusterState state = observer.setAndGetObservedState();


Lets skip this for now

Bukhtawar · 2020-11-30T16:42:18Z

...java/com/amazon/opendistroforelasticsearch/search/async/service/AsyncSearchServiceTests.java

+//import static org.mockito.Mockito.mock;
+//import static org.mockito.Mockito.when;
+//
+//public class AsyncSearchServiceTests extends ESTestCase {


Why is it commented?

…ds of search response elements

…ticsearch/asynchronous-search into async_search_service

Bukhtawar

Please address some comments i have left

Bukhtawar · 2020-12-09T08:27:09Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/plugin/AsyncSearchPlugin.java


    @Override
    public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings settings) {
-        return Collections.singletonList(new SystemIndexDescriptor(".opendistro_asynchronous_search_response",
+        return Collections.singletonList(new SystemIndexDescriptor(".opendistro_asynchroAnous_search_response",


Bukhtawar · 2020-12-09T08:29:53Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/service/AsyncSearchService.java

+            }
+            //free async search context if one exists
+            groupedDeletionListener.onResponse(asyncSearchActiveStore.freeContext(asyncSearchContextId));


Lets add an else case to avoid doubly cleaning up

Bukhtawar · 2020-12-09T08:30:32Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/service/AsyncSearchService.java

+                for (AsyncSearchActiveContext asyncSearchActiveContext : asyncSearchActiveStore.getAllContexts().values()) {
+                    AsyncSearchState stage = asyncSearchActiveContext.getAsyncSearchState();
+                    if (stage != null && (
+                            !asyncSearchActiveContext.retainedStages().contains(stage) || asyncSearchActiveContext.isExpired())) {


Replace ! with ==false

Bukhtawar · 2020-12-09T08:31:04Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/service/AsyncSearchService.java

+                    }
+                }
+            } catch (Exception e) {
+                logger.debug("Exception while reaping async search active contexts", e);


Lets improve logging, try adding async search id for every log we have for better debuggability

Bukhtawar · 2020-12-09T08:32:10Z

...om/amazon/opendistroforelasticsearch/search/async/service/active/AsyncSearchActiveStore.java

+    public boolean freeContext(AsyncSearchContextId asyncSearchContextId) {
+        AsyncSearchActiveContext asyncSearchContext = activeContexts.get(asyncSearchContextId.getId());
+        if (asyncSearchContext != null) {
+            logger.debug("Removing {} from context map", asyncSearchContextId);
+            activeContexts.remove(asyncSearchContextId.getId());
+            return true;
+        }
+        return false;
+    }


This part needs to be sync ed up with the lastest from context isAlive

Bukhtawar · 2020-12-09T08:33:46Z

...m/amazon/opendistroforelasticsearch/search/async/exception/AsyncSearchRejectedException.java

+    protected void metadataToXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.field("limit", limit);
+    }


where is it being used

https://github.com/elastic/elasticsearch/blob/master/server/src/main/java/org/elasticsearch/ElasticsearchException.java#L339-L342

Bukhtawar · 2020-12-09T08:35:37Z

...mazon/opendistroforelasticsearch/search/async/transport/TransportAsyncSearchFetchAction.java

+                                    listener.onFailure(exp);
+                                }
+                                // handle request locally if we were not able to forward the request
+                                handleRequest(asyncSearchId, request, listener);
+                            }
+                        });
+            } else {
+                handleRequest(asyncSearchId, request, listener);
+            }


Lets handle the exception only if we are unable to forward

itiyamas · 2020-12-09T10:04:05Z

...java/com/amazon/opendistroforelasticsearch/search/async/request/FetchAsyncSearchRequest.java

+
+    protected FetchAsyncSearchRequest(StreamInput in) throws IOException {
+        super(in);
+        connectionTimeout = in.readTimeValue();


Is this a required field? I think you should have a default timeout and make this optional?

Agreed. Making it optional.

itiyamas · 2020-12-09T10:06:02Z

...ava/com/amazon/opendistroforelasticsearch/search/async/request/DeleteAsyncSearchRequest.java

+/**
+ * A request to delete an async search response based on its id
+ */
+public class DeleteAsyncSearchRequest extends FetchAsyncSearchRequest<DeleteAsyncSearchRequest> {


Looks like a wierd extension name- maybe have a AsyncSearchRequest extended by both Delete and Fetch- e.g. what if the fetch request includes more params in future like pagination etc. How would that fit in with Delete?

Renaming it to AsyncSearchRoutingRequest.

itiyamas · 2020-12-09T10:07:24Z

...n/java/com/amazon/opendistroforelasticsearch/search/async/request/GetAsyncSearchRequest.java

+/**
+ * A request to fetch an async search response by id.
+ */
+public class GetAsyncSearchRequest extends FetchAsyncSearchRequest<GetAsyncSearchRequest> {


It is not clear how fetch and get are different just by looking at the class names?

Renaming it to AsyncSearchRoutingRequest.

itiyamas · 2020-12-09T10:09:35Z

...ava/com/amazon/opendistroforelasticsearch/search/async/request/SubmitAsyncSearchRequest.java

+        this.searchRequest.setCcsMinimizeRoundtrips(CCR_MINIMIZE_ROUNDTRIPS);
+        this.searchRequest.setBatchedReduceSize(DEFAULT_BATCHED_REDUCE_SIZE);
+        this.searchRequest.setPreFilterShardSize(DEFAULT_PRE_FILTER_SHARD_SIZE);
+        this.searchRequest.requestCache(DEFAULT_REQUEST_CACHE);


What if these params are already set in SearchRequest? Do you intend to override these params for async search?

Not all cases. Yes this needs a fix, we need to check values and only selectively override

itiyamas · 2020-12-09T10:16:09Z

...ava/com/amazon/opendistroforelasticsearch/search/async/request/SubmitAsyncSearchRequest.java

+    public SubmitAsyncSearchRequest(StreamInput in) throws IOException {
+        super(in);
+        this.searchRequest = new SearchRequest(in);
+        this.waitForCompletionTimeout = in.readTimeValue();


isn't this optional?

Agreed. Making it optional.

itiyamas · 2020-12-09T17:06:25Z

...main/java/com/amazon/opendistroforelasticsearch/search/async/service/AsyncSearchService.java

+                                            asyncSearchContext.getAsyncSearchId(), response);
+                                    //Intent of the lock here is to disallow ongoing migration to system index
+                                    // as if that is underway we might end up creating a new document post a DELETE was executed
+                                    asyncSearchContext.acquireContextPermit(ActionListener.wrap(
+                                            releasable -> {
+                                                //free context marks the context as DELETED
+                                                groupedDeletionListener.onResponse(
+                                                        asyncSearchActiveStore.freeContext(asyncSearchContextId));
+                                                releasable.close();
+                                            }, listener::onFailure), TimeValue.timeValueSeconds(5), "free context");
+                                },
+                                (e) -> {
+                                    asyncSearchContext.acquireContextPermit(ActionListener.wrap(
+                                            releasable -> {
+                                                groupedDeletionListener.onResponse(
+                                                        asyncSearchActiveStore.freeContext(asyncSearchContextId));
+                                                releasable.close();
+                                                //TODO introduce request timeouts to make the permit wait transparent to the client
+                                            }, listener::onFailure), TimeValue.timeValueSeconds(5), "free context");
+                                    logger.debug(() -> new ParameterizedMessage("Unable to cancel async search task {}",
+                                            asyncSearchContext.getTask()), e);


lot of duplicate code?

itiyamas · 2020-12-09T17:13:00Z

...a/com/amazon/opendistroforelasticsearch/search/async/processor/AsyncSearchPostProcessor.java

+            asyncSearchStateMachine.trigger(new SearchSuccessfulEvent(asyncSearchContext, searchResponse));
+            if (asyncSearchContext.shouldPersist()) {
+                asyncSearchStateMachine.trigger(new BeginPersistEvent(asyncSearchContext));
+                asyncSearchActiveStore.freeContext(asyncSearchContext.getContextId());


persist event execution frees the context. Why do you need to do this twice?

Agreed. this has been removed.

itiyamas · 2020-12-09T17:20:40Z

...a/com/amazon/opendistroforelasticsearch/search/async/processor/AsyncSearchPostProcessor.java

+        asyncSearchContext.acquireAllContextPermits(ActionListener.wrap(releasable -> {
+                    // check again after acquiring permit if the context has been deleted mean while
+                    if (asyncSearchContext.shouldPersist() == false) {
+                        logger.debug("Async search context {} has been moved to DELETED while waiting to acquire permits for post " +


no DELETED state as per your new logic?

itiyamas · 2020-12-09T17:31:42Z

...a/com/amazon/opendistroforelasticsearch/search/async/processor/AsyncSearchPostProcessor.java

+            asyncSearchStateMachine.trigger(new SearchFailureEvent(asyncSearchContext, exception));
+            if (asyncSearchContext.shouldPersist()) {
+                asyncSearchStateMachine.trigger(new BeginPersistEvent(asyncSearchContext));
+                asyncSearchActiveStore.freeContext(asyncSearchContext.getContextId());


Why freeing the context here?

Agreed. this has been removed.

itiyamas · 2020-12-09T17:32:22Z

...a/com/amazon/opendistroforelasticsearch/search/async/processor/AsyncSearchPostProcessor.java

+                    asyncSearchPersistenceService.storeResponse(asyncSearchContext.getAsyncSearchId(),
+                            persistenceModel, ActionListener.runAfter(ActionListener.wrap(
+                                    (indexResponse) -> {
+                                        //Mark any dangling reference as PERSISTED and cleaning it up from the IN_MEMORY context


Why do you call it dangling?

Concurrent get or delete requests may be holding reference to the active context.

…ticsearch/asynchronous-search into async_search_service

itiyamas · 2020-12-11T14:25:11Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

+        }
+        return true;
+    }
+
    @Override
    public void close() {


synchronized?

itiyamas · 2020-12-11T14:27:30Z

.../amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveContext.java

@@ -59,6 +61,7 @@
    private final AtomicBoolean completed;
    private final SetOnce<Exception> error;
    private final SetOnce<SearchResponse> searchResponse;
+    private final AtomicBoolean closed;
    private final AsyncSearchContextPermits asyncSearchContextPermits;

    public AsyncSearchActiveContext(AsyncSearchContextId asyncSearchContextId, String nodeId,


Can this be executing even after close is called? If not, then you need to check for closed flag in every method?

Yes. Have added this check in every method.

itiyamas · 2020-12-11T14:29:53Z

src/main/java/com/amazon/opendistroforelasticsearch/search/async/plugin/AsyncSearchPlugin.java

    }

-    private AsyncSearchStateMachine getAsyncSearchStateMachineDefinition() {


Where has this gone now?

AsyncSearchStateMachine is being initialized in the AsyncSearchService

…Supplier to System.CurrentTimeMillis

…ion tests

…ource clean up in integration test.

itiyamas

As discussed, please be careful about ResourceNotFound exception. It might break some consistency guarantees of your API for concurrent DELETE and GET calls. You might want to throw NotAvailable instead of NotFound to preserve strong consistency.

getsaurabh02

Awesome job on the tests! looks pretty much in the good shape.

eirsep added 5 commits November 23, 2020 17:26

async search service. async search submit, get,delete transport actions.

6c56c5f

fix checkstyle violations

989f153

delete .idea folder

173f0f4

async search post processor fix

716c4cf

async search post processor fix

86073c2

eirsep requested review from itiyamas and getsaurabh02 November 24, 2020 04:28

Bukhtawar changed the title ~~async search service. async search submit, get,delete transport actions.~~ Service layer and Transport Handlers Nov 24, 2020

eirsep added 6 commits November 24, 2020 15:10

clear partial result holder on search completion. add java docs

79a3eb6

persistence service tests. integration tests set up. isInitialized fi…

8da4ba4

…x in partial result holder

match field name in persistence context to response

a3b3199

serialization change in persistence layer. store error, response as b…

689205c

…inary

test assert fix

8d8db51

assert message on failure to debug error

a2ea670

Bukhtawar mentioned this pull request Nov 30, 2020

Asynchronous Search component breakdown #3

Closed

8 tasks

eirsep added 2 commits November 30, 2020 21:24

async search service java docs. package restructuring

20bea89

await on correct latch in test

bbc1aed

Bukhtawar reviewed Nov 30, 2020

View reviewed changes

eirsep and others added 10 commits November 30, 2020 22:35

countdown correct latch in test

7a76cde

minor fixes to address review comments.

f7bde86

async search service tests

2b29782

async search query tests added to verify persistence of different kin…

63fd1b0

…ds of search response elements

async search state machine changes. persisting stage added.

a072fc7

request routing layer fix

ace1ae1

fix indices clearance

35ebaca

Task cancellation and Tests fixup

835fa23

test renamed. minor fixes in tests

39a97d4

Merge branch 'async_search_service' of github.com:opendistro-for-elas…

14d447e

…ticsearch/asynchronous-search into async_search_service

Bukhtawar suggested changes Dec 9, 2020

View reviewed changes

itiyamas suggested changes Dec 9, 2020

View reviewed changes

eirsep and others added 9 commits December 11, 2020 09:46

Merge branch 'async_search_service' of github.com:opendistro-for-elas…

0d4a2d4

…ticsearch/asynchronous-search into async_search_service

Simplified free Active Context

a316727

Simplified free Active Context

c6d26b4

Merge branch 'async_search_service' of github.com:opendistro-for-elas…

15d019d

…ticsearch/asynchronous-search into async_search_service

Exception translation and Async search service- plugin changes

32c1490

Merge branch 'async_search_service' of github.com:opendistro-for-elas…

0cc0ca7

…ticsearch/asynchronous-search into async_search_service

fix checkstyle violations

dc48588

fix service tests. set refresh flag in get request

7c85af3

randomize keepOnCompletion flag in async search serice test

155d443

itiyamas reviewed Dec 11, 2020

View reviewed changes

eirsep and others added 9 commits December 12, 2020 09:42

changes to async search service and active context tests. change long…

96f92d6

…Supplier to System.CurrentTimeMillis

Multi-threaded tests and bug fixes

847b785

Multi-threaded tests fixes and rename DELETED to CLOSED

fc27075

Persistence delete expiration time tests

e457013

async search state machine tests. submit async search request validat…

83a829a

…ion tests

check response doc removed from index in async search deletion IT

c1a116b

Async search cancellation changes

369d141

minor fix in state machine tests

48507ae

erroneous test omitted

59a9e3f

Bukhtawar requested a review from rramachand21 December 14, 2020 06:21

add synchronization block to isAlive. assert active and persisted res…

7e1d2e7

…ource clean up in integration test.

itiyamas approved these changes Dec 14, 2020

View reviewed changes

Bukhtawar requested a review from getsaurabh02 December 14, 2020 07:46

revert synchronization block in isAlive() check

859154e

getsaurabh02 approved these changes Dec 14, 2020

View reviewed changes

Exception Translation changes

57df6b7

Bukhtawar merged commit 24440c7 into master Dec 14, 2020

Bukhtawar deleted the async_search_service branch January 2, 2021 18:34

		}

		private AsyncSearchStateMachine getAsyncSearchStateMachineDefinition() {

Service layer and Transport Handlers #8

Service layer and Transport Handlers #8

Conversation

eirsep commented Nov 23, 2020 • edited Loading

Issue #3

Bukhtawar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Bukhtawar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

itiyamas left a comment

Choose a reason for hiding this comment

getsaurabh02 left a comment

Choose a reason for hiding this comment

eirsep commented Nov 23, 2020 •

edited

Loading