Skip to content

Commit f207d15

Browse files
arteamarteam
committed
Bring back original TransportGetFromTranslogAction and TransportShardMultiGetFomTranslogAction
1 parent 47b5083 commit f207d15

File tree

2 files changed

+56
-76
lines changed

2 files changed

+56
-76
lines changed

‎server/src/main/java/org/elasticsearch/action/get/TransportGetFromTranslogAction.java‎

Lines changed: 16 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -58,9 +58,23 @@ public TransportGetFromTranslogAction(TransportService transportService, Indices
5858

5959
@Override
6060
protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
61-
IndexShard indexShard = getIndexShard(indicesService, request);
61+
final GetRequest getRequest = request.getRequest();
62+
final ShardId shardId = request.shardId();
63+
IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
64+
IndexShard indexShard = indexService.getShard(shardId.id());
65+
assert indexShard.routingEntry().isPromotableToPrimary() : "not an indexing shard" + indexShard.routingEntry();
66+
assert getRequest.realtime();
6267
ActionListener.completeWith(listener, () -> {
63-
var result = getResult(indexShard, request.getRequest());
68+
var result = indexShard.getService()
69+
.getFromTranslog(
70+
getRequest.id(),
71+
getRequest.storedFields(),
72+
getRequest.realtime(),
73+
getRequest.version(),
74+
getRequest.versionType(),
75+
getRequest.fetchSourceContext(),
76+
getRequest.isForceSyntheticSource()
77+
);
6478
long segmentGeneration = -1;
6579
if (result == null) {
6680
Engine engine = indexShard.getEngineOrNull();
@@ -73,28 +87,6 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
7387
});
7488
}
7589

76-
public static IndexShard getIndexShard(IndicesService indicesService, Request request) {
77-
final ShardId shardId = request.shardId();
78-
IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
79-
IndexShard indexShard = indexService.getShard(shardId.id());
80-
assert indexShard.routingEntry().isPromotableToPrimary() : "not an indexing shard" + indexShard.routingEntry();
81-
assert request.getRequest().realtime();
82-
return indexShard;
83-
}
84-
85-
public static GetResult getResult(IndexShard indexShard, GetRequest getRequest) throws IOException {
86-
return indexShard.getService()
87-
.getFromTranslog(
88-
getRequest.id(),
89-
getRequest.storedFields(),
90-
getRequest.realtime(),
91-
getRequest.version(),
92-
getRequest.versionType(),
93-
getRequest.fetchSourceContext(),
94-
getRequest.isForceSyntheticSource()
95-
);
96-
}
97-
9890
public static class Request extends ActionRequest implements IndicesRequest {
9991

10092
private final GetRequest getRequest;

‎server/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetFomTranslogAction.java‎

Lines changed: 40 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121
import org.elasticsearch.common.Strings;
2222
import org.elasticsearch.common.io.stream.StreamInput;
2323
import org.elasticsearch.common.io.stream.StreamOutput;
24-
import org.elasticsearch.core.Tuple;
2524
import org.elasticsearch.index.IndexService;
2625
import org.elasticsearch.index.engine.Engine;
2726
import org.elasticsearch.index.engine.InternalEngine;
@@ -58,68 +57,57 @@ protected TransportShardMultiGetFomTranslogAction(
5857
@Override
5958
protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
6059
var multiGetShardRequest = request.getMultiGetShardRequest();
61-
IndexShard indexShard = getIndexShard(indicesService, request);
60+
var shardId = request.getShardId();
61+
IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
62+
IndexShard indexShard = indexService.getShard(shardId.id());
63+
assert indexShard.routingEntry().isPromotableToPrimary() : "not an indexing shard" + indexShard.routingEntry();
64+
assert multiGetShardRequest.realtime();
6265
ActionListener.completeWith(listener, () -> {
63-
Tuple<MultiGetShardResponse, Boolean> multiGetShardResponse = getResponse(multiGetShardRequest, indexShard);
66+
var multiGetShardResponse = new MultiGetShardResponse();
67+
var someItemsNotFoundInTranslog = false;
68+
for (int i = 0; i < multiGetShardRequest.locations.size(); i++) {
69+
var item = multiGetShardRequest.items.get(i);
70+
try {
71+
var result = indexShard.getService()
72+
.getFromTranslog(
73+
item.id(),
74+
item.storedFields(),
75+
multiGetShardRequest.realtime(),
76+
item.version(),
77+
item.versionType(),
78+
item.fetchSourceContext(),
79+
multiGetShardRequest.isForceSyntheticSource()
80+
);
81+
GetResponse getResponse = null;
82+
if (result == null) {
83+
someItemsNotFoundInTranslog = true;
84+
} else {
85+
getResponse = new GetResponse(result);
86+
}
87+
multiGetShardResponse.add(multiGetShardRequest.locations.get(i), getResponse);
88+
} catch (RuntimeException | IOException e) {
89+
if (TransportActions.isShardNotAvailableException(e)) {
90+
throw e;
91+
}
92+
logger.debug("failed to execute multi_get_from_translog for {}[id={}]: {}", shardId, item.id(), e);
93+
multiGetShardResponse.add(
94+
multiGetShardRequest.locations.get(i),
95+
new MultiGetResponse.Failure(multiGetShardRequest.index(), item.id(), e)
96+
);
97+
}
98+
}
6499
long segmentGeneration = -1;
65-
if (multiGetShardResponse.v2()) {
100+
if (someItemsNotFoundInTranslog) {
66101
Engine engine = indexShard.getEngineOrNull();
67102
if (engine == null) {
68103
throw new AlreadyClosedException("engine closed");
69104
}
70105
segmentGeneration = ((InternalEngine) engine).getLastUnsafeSegmentGenerationForGets();
71106
}
72-
return new Response(multiGetShardResponse.v1(), indexShard.getOperationPrimaryTerm(), segmentGeneration);
107+
return new Response(multiGetShardResponse, indexShard.getOperationPrimaryTerm(), segmentGeneration);
73108
});
74109
}
75110

76-
public static IndexShard getIndexShard(IndicesService indicesService, Request request) {
77-
var shardId = request.getShardId();
78-
IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
79-
IndexShard indexShard = indexService.getShard(shardId.id());
80-
assert indexShard.routingEntry().isPromotableToPrimary() : "not an indexing shard" + indexShard.routingEntry();
81-
assert request.getMultiGetShardRequest().realtime();
82-
return indexShard;
83-
}
84-
85-
public static Tuple<MultiGetShardResponse, Boolean> getResponse(MultiGetShardRequest multiGetShardRequest, IndexShard indexShard)
86-
throws IOException {
87-
var multiGetShardResponse = new MultiGetShardResponse();
88-
var someItemsNotFoundInTranslog = false;
89-
for (int i = 0; i < multiGetShardRequest.locations.size(); i++) {
90-
var item = multiGetShardRequest.items.get(i);
91-
try {
92-
var result = indexShard.getService()
93-
.getFromTranslog(
94-
item.id(),
95-
item.storedFields(),
96-
multiGetShardRequest.realtime(),
97-
item.version(),
98-
item.versionType(),
99-
item.fetchSourceContext(),
100-
multiGetShardRequest.isForceSyntheticSource()
101-
);
102-
GetResponse getResponse = null;
103-
if (result == null) {
104-
someItemsNotFoundInTranslog = true;
105-
} else {
106-
getResponse = new GetResponse(result);
107-
}
108-
multiGetShardResponse.add(multiGetShardRequest.locations.get(i), getResponse);
109-
} catch (RuntimeException | IOException e) {
110-
if (TransportActions.isShardNotAvailableException(e)) {
111-
throw e;
112-
}
113-
logger.debug("failed to execute multi_get_from_translog for {}[id={}]: {}", multiGetShardRequest.shardId(), item.id(), e);
114-
multiGetShardResponse.add(
115-
multiGetShardRequest.locations.get(i),
116-
new MultiGetResponse.Failure(multiGetShardRequest.index(), item.id(), e)
117-
);
118-
}
119-
}
120-
return Tuple.tuple(multiGetShardResponse, someItemsNotFoundInTranslog);
121-
}
122-
123111
public static class Request extends ActionRequest {
124112

125113
private final MultiGetShardRequest multiGetShardRequest;

0 commit comments

Comments
 (0)