Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import java.util.List;
import java.util.Map;

import org.opensearch.ExceptionsHelper;
import org.opensearch.OpenSearchStatusException;
import org.opensearch.action.DocWriteRequest;
import org.opensearch.action.admin.cluster.storedscripts.GetStoredScriptRequest;
Expand Down Expand Up @@ -42,6 +43,7 @@
import org.opensearch.core.xcontent.ToXContentObject;
import org.opensearch.core.xcontent.XContentParser;
import org.opensearch.index.IndexNotFoundException;
import org.opensearch.index.engine.DocumentMissingException;
import org.opensearch.index.engine.VersionConflictEngineException;
import org.opensearch.index.query.MatchAllQueryBuilder;
import org.opensearch.ml.common.CommonValue;
Expand Down Expand Up @@ -202,7 +204,7 @@ private void storeTemplateInIndex(AgenticSearchTemplate template, ActionListener
log.info("Registered agentic search template: {}", template.getTemplateId());
listener.onResponse(true);
}, e -> {
if (e instanceof VersionConflictEngineException) {
if (ExceptionsHelper.unwrapCause(e) instanceof VersionConflictEngineException) {
listener
.onFailure(
new OpenSearchStatusException(
Expand Down Expand Up @@ -474,7 +476,7 @@ private void fetchFlattenedMapping(String index, ActionListener<List<String>> li
listener.onFailure(e);
}
}, e -> {
if (e instanceof IndexNotFoundException) {
if (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) {
listener.onFailure(new OpenSearchStatusException("Index does not exist: " + index, RestStatus.BAD_REQUEST));
} else {
listener.onFailure(e);
Expand Down Expand Up @@ -547,7 +549,7 @@ public void getTemplate(String templateId, ActionListener<AgenticSearchTemplate>
wrapped.onFailure(e);
}
}, e -> {
if (e instanceof IndexNotFoundException) {
if (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) {
wrapped.onFailure(new OpenSearchStatusException(NOT_FOUND_ERROR + templateId, RestStatus.NOT_FOUND));
} else {
wrapped.onFailure(e);
Expand Down Expand Up @@ -575,7 +577,7 @@ public void listTemplates(int from, int size, ActionListener<MLListResult> liste
wrapped.onFailure(e);
}
}, e -> {
if (e instanceof IndexNotFoundException) {
if (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) {
wrapped.onResponse(new MLListResult(new ArrayList<>(), 0));
} else {
wrapped.onFailure(e);
Expand All @@ -598,7 +600,7 @@ public void deleteTemplate(String templateId, ActionListener<Boolean> listener)
}
wrapped.onResponse(true);
}, e -> {
if (e instanceof IndexNotFoundException) {
if (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) {
wrapped.onFailure(new OpenSearchStatusException(NOT_FOUND_ERROR + templateId, RestStatus.NOT_FOUND));
} else {
wrapped.onFailure(e);
Expand Down Expand Up @@ -653,7 +655,7 @@ public void updateTemplate(String templateId, AgenticSearchTemplate patch, Actio
writeTemplatePatch(templateId, patch, seqNo, primaryTerm, wrapped);
}, wrapped::onFailure));
}, e -> {
if (e instanceof IndexNotFoundException) {
if (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) {
wrapped.onFailure(new OpenSearchStatusException(NOT_FOUND_ERROR + templateId, RestStatus.NOT_FOUND));
} else {
wrapped.onFailure(e);
Expand Down Expand Up @@ -686,7 +688,9 @@ private void writeTemplatePatch(
updateRequest.setIfSeqNo(seqNo).setIfPrimaryTerm(primaryTerm);
}
client.update(updateRequest, ActionListener.wrap(listener::onResponse, e -> {
if (e instanceof org.opensearch.index.engine.DocumentMissingException || e instanceof IndexNotFoundException) {
// A remote shard failure arrives wrapped in RemoteTransportException.
Throwable cause = ExceptionsHelper.unwrapCause(e);
if (cause instanceof DocumentMissingException || cause instanceof IndexNotFoundException) {
listener.onFailure(new OpenSearchStatusException(NOT_FOUND_ERROR + templateId, RestStatus.NOT_FOUND));
} else {
listener.onFailure(e);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@
import org.opensearch.search.SearchHits;
import org.opensearch.test.OpenSearchTestCase;
import org.opensearch.threadpool.ThreadPool;
import org.opensearch.transport.RemoteTransportException;
import org.opensearch.transport.client.AdminClient;
import org.opensearch.transport.client.Client;
import org.opensearch.transport.client.ClusterAdminClient;
Expand Down Expand Up @@ -210,6 +211,29 @@ public void register_existingId_failsConflict() {
assertStatus(listener, RestStatus.CONFLICT, "already exists");
}

@Test
public void register_existingId_remoteWrapped_failsConflict() {
stubStoredScript(TEMPLATE_BODY);
stubIndexMapping(ImmutableMap.of("properties", ImmutableMap.of("title", ImmutableMap.of("type", "text"))));
stubRenderSucceeds();
doAnswer((Answer<Void>) inv -> {
ActionListener<Boolean> l = inv.getArgument(1);
l.onResponse(true);
return null;
}).when(mlIndicesHandler).initMLIndexIfAbsent(any(), any());
doAnswer((Answer<Void>) inv -> {
ActionListener<IndexResponse> l = inv.getArgument(1);
l.onFailure(remote(new org.opensearch.index.engine.VersionConflictEngineException(null, "tmpl", "already exists")));
return null;
}).when(client).index(any(IndexRequest.class), any());

@SuppressWarnings("unchecked")
ActionListener<AgenticSearchTemplate> listener = mock(ActionListener.class);
service.register("tmpl", "my-index", "desc", null, null, listener);

assertStatus(listener, RestStatus.CONFLICT, "already exists");
}

@Test
public void register_setsOpTypeCreate() throws Exception {
stubStoredScript(TEMPLATE_BODY);
Expand Down Expand Up @@ -541,6 +565,15 @@ public void getTemplate_indexNotFound_failsNotFound() {
assertStatus(listener, RestStatus.NOT_FOUND, "not found");
}

@Test
public void getTemplate_indexNotFound_remoteWrapped_failsNotFound() {
stubGetFailure(remote(new IndexNotFoundException("idx")));
@SuppressWarnings("unchecked")
ActionListener<AgenticSearchTemplate> listener = mock(ActionListener.class);
service.getTemplate("tmpl", listener);
assertStatus(listener, RestStatus.NOT_FOUND, "not found");
}

@Test
public void getTemplate_otherException_propagates() {
stubGetFailure(new RuntimeException("boom"));
Expand Down Expand Up @@ -623,6 +656,15 @@ public void deleteTemplate_indexNotFound_failsNotFound() {
assertStatus(listener, RestStatus.NOT_FOUND, "not found");
}

@Test
public void deleteTemplate_indexNotFound_remoteWrapped_failsNotFound() {
stubDelete(null, remote(new IndexNotFoundException("idx")));
@SuppressWarnings("unchecked")
ActionListener<Boolean> listener = mock(ActionListener.class);
service.deleteTemplate("tmpl", listener);
assertStatus(listener, RestStatus.NOT_FOUND, "not found");
}

@Test
public void deleteTemplate_otherException_propagates() {
stubDelete(null, new RuntimeException("del boom"));
Expand Down Expand Up @@ -651,8 +693,44 @@ public void update_noParamSchema_documentMissing_failsNotFound() {
assertStatus(listener, RestStatus.NOT_FOUND, "not found");
}

@Test
public void update_noParamSchema_documentMissing_remoteWrapped_failsNotFound() {
// Production shape: the shard-level failure arrives wrapped in RemoteTransportException.
AgenticSearchTemplate patch = AgenticSearchTemplate.builder().templateId("tmpl").description("d").build();
doAnswer((Answer<Void>) inv -> {
ActionListener<UpdateResponse> l = inv.getArgument(1);
l.onFailure(remote(new org.opensearch.index.engine.DocumentMissingException(null, "tmpl")));
return null;
}).when(client).update(any(UpdateRequest.class), any());

@SuppressWarnings("unchecked")
ActionListener<UpdateResponse> listener = mock(ActionListener.class);
service.updateTemplate("tmpl", patch, listener);
assertStatus(listener, RestStatus.NOT_FOUND, "Agentic search template not found: tmpl");
}

@Test
public void update_noParamSchema_otherException_propagatesOriginal() {
AgenticSearchTemplate patch = AgenticSearchTemplate.builder().templateId("tmpl").description("d").build();
RemoteTransportException failure = remote(new RuntimeException("update boom"));
doAnswer((Answer<Void>) inv -> {
ActionListener<UpdateResponse> l = inv.getArgument(1);
l.onFailure(failure);
return null;
}).when(client).update(any(UpdateRequest.class), any());

@SuppressWarnings("unchecked")
ActionListener<UpdateResponse> listener = mock(ActionListener.class);
service.updateTemplate("tmpl", patch, listener);
verify(listener).onFailure(failure);
}

// ---- helpers -----------------------------------------------------------

private static RemoteTransportException remote(Exception cause) {
return new RemoteTransportException("[node][127.0.0.1:9300][indices:data/write]", cause);
}

// A stored template doc as persisted in the system index (parsed by get/list).
private static final String TEMPLATE_DOC = "{\"template_id\":\"tmpl\",\"index_binding\":\"my-index\","
+ "\"param_schema\":{\"lex_query\":{\"type\":\"string\",\"required\":true}}}";
Expand Down
Loading