Skip to content
Merged
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 @@ -20,17 +20,23 @@

import org.elasticsearch.action.Action;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.io.stream.Writeable;

public class NoopSearchAction extends Action<SearchResponse> {
public static final NoopSearchAction INSTANCE = new NoopSearchAction();
public static final String NAME = "mock:data/read/search";

public NoopSearchAction() {
private NoopSearchAction() {
super(NAME);
}

@Override
public SearchResponse newResponse() {
return new SearchResponse();
throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
}

@Override
public Writeable.Reader<SearchResponse> getResponseReader() {
return SearchResponse::new;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,7 @@ public static class Item implements Writeable {

private Item(StreamInput in) throws IOException {
if (in.readBoolean()) {
this.response = new SearchTemplateResponse();
response.readFrom(in);
this.response = new SearchTemplateResponse(in);
this.exception = null;
} else {
exception = in.readException();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.script.mustache;

import org.elasticsearch.action.Action;
import org.elasticsearch.common.io.stream.Writeable;

public class SearchTemplateAction extends Action<SearchTemplateResponse> {

Expand All @@ -32,6 +33,11 @@ private SearchTemplateAction() {

@Override
public SearchTemplateResponse newResponse() {
return new SearchTemplateResponse();
throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
}

@Override
public Writeable.Reader<SearchTemplateResponse> getResponseReader() {
return SearchTemplateResponse::new;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,12 @@ public class SearchTemplateResponse extends ActionResponse implements StatusToXC
SearchTemplateResponse() {
}

SearchTemplateResponse(StreamInput in) throws IOException {
Copy link
Contributor

@mayya-sharipova mayya-sharipova May 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if we can remove an empty constructor and make all member variables final here? We may need to change a little fromXContent and have another constructor.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is possible yet it requires some work on how SearchTemplateResponse is used, as source and response are generally set after creation. I would prefer to keep this out of this PR.

super(in);
source = in.readOptionalBytesReference();
response = in.readOptionalWriteable(SearchResponse::new);
}

public BytesReference getSource() {
return source;
}
Expand Down Expand Up @@ -81,10 +87,8 @@ public void writeTo(StreamOutput out) throws IOException {
}

@Override
public void readFrom(StreamInput in) throws IOException {
super.readFrom(in);
source = in.readOptionalBytesReference();
response = in.readOptionalStreamable(SearchResponse::new);
public void readFrom(StreamInput in) {
throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
}

public static SearchTemplateResponse fromXContent(XContentParser parser) throws IOException {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -70,8 +70,7 @@ public Item(SearchResponse response, Exception exception) {

Item(StreamInput in) throws IOException{
if (in.readBoolean()) {
this.response = new SearchResponse();
this.response.readFrom(in);
this.response = new SearchResponse(in);
this.exception = null;
} else {
this.exception = in.readException();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.search;

import org.elasticsearch.action.Action;
import org.elasticsearch.common.io.stream.Writeable;

public class SearchAction extends Action<SearchResponse> {

Expand All @@ -32,6 +33,11 @@ private SearchAction() {

@Override
public SearchResponse newResponse() {
return new SearchResponse();
throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
}

@Override
public Writeable.Reader<SearchResponse> getResponseReader() {
return SearchResponse::new;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -66,23 +66,33 @@ public class SearchResponse extends ActionResponse implements StatusToXContentOb
private static final ParseField TERMINATED_EARLY = new ParseField("terminated_early");
private static final ParseField NUM_REDUCE_PHASES = new ParseField("num_reduce_phases");

private SearchResponseSections internalResponse;

private String scrollId;

private int totalShards;

private int successfulShards;

private int skippedShards;

private ShardSearchFailure[] shardFailures;

private Clusters clusters;

private long tookInMillis;

public SearchResponse() {
private final SearchResponseSections internalResponse;
private final String scrollId;
private final int totalShards;
private final int successfulShards;
private final int skippedShards;
private final ShardSearchFailure[] shardFailures;
private final Clusters clusters;
private final long tookInMillis;

public SearchResponse(StreamInput in) throws IOException {
Copy link
Contributor

@mayya-sharipova mayya-sharipova May 7, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we also remove an empty constructor and make all member variables final here?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes this is not so hard so I am making this change.

super(in);
internalResponse = new InternalSearchResponse(in);
totalShards = in.readVInt();
successfulShards = in.readVInt();
int size = in.readVInt();
if (size == 0) {
shardFailures = ShardSearchFailure.EMPTY_ARRAY;
} else {
shardFailures = new ShardSearchFailure[size];
for (int i = 0; i < shardFailures.length; i++) {
shardFailures[i] = readShardSearchFailure(in);
}
}
clusters = new Clusters(in);
scrollId = in.readOptionalString();
tookInMillis = in.readVLong();
skippedShards = in.readVInt();
}

public SearchResponse(SearchResponseSections internalResponse, String scrollId, int totalShards, int successfulShards,
Expand Down Expand Up @@ -193,10 +203,6 @@ public String getScrollId() {
return scrollId;
}

public void scrollId(String scrollId) {
this.scrollId = scrollId;
}

/**
* If profiling was enabled, this returns an object containing the profile results from
* each shard. If profiling was not enabled, this will return null
Expand Down Expand Up @@ -355,24 +361,8 @@ static SearchResponse innerFromXContent(XContentParser parser) throws IOExceptio
}

@Override
public void readFrom(StreamInput in) throws IOException {
super.readFrom(in);
internalResponse = new InternalSearchResponse(in);
totalShards = in.readVInt();
successfulShards = in.readVInt();
int size = in.readVInt();
if (size == 0) {
shardFailures = ShardSearchFailure.EMPTY_ARRAY;
} else {
shardFailures = new ShardSearchFailure[size];
for (int i = 0; i < shardFailures.length; i++) {
shardFailures[i] = readShardSearchFailure(in);
}
}
clusters = new Clusters(in);
scrollId = in.readOptionalString();
tookInMillis = in.readVLong();
skippedShards = in.readVInt();
public void readFrom(StreamInput in) {
throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.elasticsearch.action.search;

import org.elasticsearch.action.Action;
import org.elasticsearch.common.io.stream.Writeable;

public class SearchScrollAction extends Action<SearchResponse> {

Expand All @@ -32,6 +33,11 @@ private SearchScrollAction() {

@Override
public SearchResponse newResponse() {
return new SearchResponse();
throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
}

@Override
public Writeable.Reader<SearchResponse> getResponseReader() {
return SearchResponse::new;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.AtomicArray;
import org.elasticsearch.search.internal.InternalSearchResponse;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.tasks.TaskManager;
import org.elasticsearch.test.ESTestCase;
Expand Down Expand Up @@ -154,7 +155,8 @@ public void search(final SearchRequest request, final ActionListener<SearchRespo
requests.add(request);
commonExecutor.execute(() -> {
counter.decrementAndGet();
listener.onResponse(new SearchResponse());
listener.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, 0, 0L,
ShardSearchFailure.EMPTY_ARRAY, SearchResponse.Clusters.EMPTY));
});
}
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.search.SearchPhaseResult;
import org.elasticsearch.search.internal.AliasFilter;
import org.elasticsearch.search.internal.InternalSearchResponse;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.transport.Transport;
import org.elasticsearch.transport.TransportException;
Expand Down Expand Up @@ -406,13 +407,17 @@ static GroupShardsIterator<SearchShardIterator> getShardsIter(String index, Orig
}

public static class TestSearchResponse extends SearchResponse {
public final Set<ShardId> queried = new HashSet<>();
final Set<ShardId> queried = new HashSet<>();

TestSearchResponse() {
super(InternalSearchResponse.empty(), null, 0, 0, 0, 0L, ShardSearchFailure.EMPTY_ARRAY, Clusters.EMPTY);
}
}

public static class TestSearchPhaseResult extends SearchPhaseResult {
final DiscoveryNode node;

public TestSearchPhaseResult(long id, DiscoveryNode node) {
TestSearchPhaseResult(long id, DiscoveryNode node) {
this.requestId = id;
this.node = node;
}
Expand All @@ -427,7 +432,7 @@ public static final class MockConnection implements Transport.Connection {

private final DiscoveryNode node;

public MockConnection(DiscoveryNode node) {
MockConnection(DiscoveryNode node) {
this.node = node;
}

Expand All @@ -438,7 +443,7 @@ public DiscoveryNode getNode() {

@Override
public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
throws IOException, TransportException {
throws TransportException {
throw new UnsupportedOperationException();
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ public void testToXContent() {

public void testSerialization() throws IOException {
SearchResponse searchResponse = createTestItem(false);
SearchResponse deserialized = copyStreamable(searchResponse, namedWriteableRegistry, SearchResponse::new, Version.CURRENT);
SearchResponse deserialized = copyWriteable(searchResponse, namedWriteableRegistry, SearchResponse::new, Version.CURRENT);
if (searchResponse.getHits().getTotalHits() == null) {
assertNull(deserialized.getHits().getTotalHits());
} else {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.elasticsearch.common.Randomness;
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.search.internal.InternalSearchResponse;
import org.elasticsearch.tasks.TaskManager;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.threadpool.TestThreadPool;
Expand Down Expand Up @@ -117,7 +118,8 @@ public void search(final SearchRequest request, final ActionListener<SearchRespo
final ExecutorService executorService = rarely() ? rarelyExecutor : commonExecutor;
executorService.execute(() -> {
counter.decrementAndGet();
listener.onResponse(new SearchResponse());
listener.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, 0, 0L,
ShardSearchFailure.EMPTY_ARRAY, SearchResponse.Clusters.EMPTY));
});
}
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.ElasticsearchClient;
import org.elasticsearch.common.io.stream.Writeable;

public class RollupSearchAction extends Action<SearchResponse> {

Expand All @@ -22,16 +23,17 @@ private RollupSearchAction() {

@Override
public SearchResponse newResponse() {
return new SearchResponse();
throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
}

@Override
public Writeable.Reader<SearchResponse> getResponseReader() {
return SearchResponse::new;
}

public static class RequestBuilder extends ActionRequestBuilder<SearchRequest, SearchResponse> {
public RequestBuilder(ElasticsearchClient client, SearchRequest searchRequest) {
super(client, INSTANCE, searchRequest);
}

RequestBuilder(ElasticsearchClient client) {
super(client, INSTANCE, new SearchRequest());
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,13 @@
import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.action.support.PlainActionFuture;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.collect.MapBuilder;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.search.internal.InternalSearchResponse;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
Expand Down Expand Up @@ -232,7 +234,8 @@ public void testExecuteWithHeadersNoHeaders() {
when(client.threadPool()).thenReturn(threadPool);

PlainActionFuture<SearchResponse> searchFuture = PlainActionFuture.newFuture();
searchFuture.onResponse(new SearchResponse());
searchFuture.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, 0, 0L, ShardSearchFailure.EMPTY_ARRAY,
SearchResponse.Clusters.EMPTY));
when(client.search(any())).thenReturn(searchFuture);
assertExecutionWithOrigin(Collections.emptyMap(), client);
}
Expand All @@ -245,7 +248,8 @@ public void testExecuteWithHeaders() {
when(client.threadPool()).thenReturn(threadPool);

PlainActionFuture<SearchResponse> searchFuture = PlainActionFuture.newFuture();
searchFuture.onResponse(new SearchResponse());
searchFuture.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, 0, 0L, ShardSearchFailure.EMPTY_ARRAY,
SearchResponse.Clusters.EMPTY));
when(client.search(any())).thenReturn(searchFuture);
Map<String, String> headers = MapBuilder.<String, String> newMapBuilder().put(AuthenticationField.AUTHENTICATION_KEY, "anything")
.put(AuthenticationServiceField.RUN_AS_USER_HEADER, "anything").map();
Expand All @@ -265,7 +269,8 @@ public void testExecuteWithHeadersNoSecurityHeaders() {
when(client.threadPool()).thenReturn(threadPool);

PlainActionFuture<SearchResponse> searchFuture = PlainActionFuture.newFuture();
searchFuture.onResponse(new SearchResponse());
searchFuture.onResponse(new SearchResponse(InternalSearchResponse.empty(), null, 0, 0, 0, 0L, ShardSearchFailure.EMPTY_ARRAY,
SearchResponse.Clusters.EMPTY));
when(client.search(any())).thenReturn(searchFuture);
Map<String, String> unrelatedHeaders = MapBuilder.<String, String> newMapBuilder().put(randomAlphaOfLength(10), "anything").map();

Expand Down