Skip to content

Commit 28660be

Browse files
jasontedorjasontedor
authored
Fix resync request serialization
This commit addresses a subtle bug in the serialization routine for resync requests. The problem here is that Translog.Operation#readType is not compatible with the implementations of Translog.Operation#writeTo. Unfortunately, this issue prevents primary-replica from succeeding, issues which we will address in follow-ups. Relates #27418
1 parent 4e04f95 commit 28660be

File tree

8 files changed

+166
-43
lines changed

8 files changed

+166
-43
lines changed

‎core/src/main/java/org/elasticsearch/action/resync/ResyncReplicationRequest.java‎

Lines changed: 36 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,42 +18,68 @@
1818
*/
1919
package org.elasticsearch.action.resync;
2020

21+
import org.elasticsearch.Version;
2122
import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
2223
import org.elasticsearch.common.io.stream.StreamInput;
2324
import org.elasticsearch.common.io.stream.StreamOutput;
2425
import org.elasticsearch.index.shard.ShardId;
2526
import org.elasticsearch.index.translog.Translog;
2627

2728
import java.io.IOException;
28-
import java.util.List;
29+
import java.util.Arrays;
2930

31+
/**
32+
* Represents a batch of operations sent from the primary to its replicas during the primary-replica resync.
33+
*/
3034
public final class ResyncReplicationRequest extends ReplicatedWriteRequest<ResyncReplicationRequest> {
3135

32-
private List<Translog.Operation> operations;
36+
private Translog.Operation[] operations;
3337

3438
ResyncReplicationRequest() {
3539
super();
3640
}
3741

38-
public ResyncReplicationRequest(ShardId shardId, List<Translog.Operation> operations) {
42+
public ResyncReplicationRequest(final ShardId shardId, final Translog.Operation[] operations) {
3943
super(shardId);
4044
this.operations = operations;
4145
}
4246

43-
public List<Translog.Operation> getOperations() {
47+
public Translog.Operation[] getOperations() {
4448
return operations;
4549
}
4650

4751
@Override
48-
public void readFrom(StreamInput in) throws IOException {
52+
public void readFrom(final StreamInput in) throws IOException {
53+
if (in.getVersion().equals(Version.V_6_0_0)) {
54+
/*
55+
* Resync replication request serialization was broken in 6.0.0 due to the elements of the stream not being prefixed with a
56+
* byte indicating the type of the operation.
57+
*/
58+
// TODO: remove this check in 8.0.0 which provides no BWC guarantees with 6.x.
59+
assert Version.CURRENT.major <= 7;
60+
throw new IllegalStateException("resync replication request serialization is broken in 6.0.0");
61+
}
4962
super.readFrom(in);
50-
operations = in.readList(Translog.Operation::readType);
63+
operations = in.readArray(Translog.Operation::readOperation, Translog.Operation[]::new);
5164
}
5265

5366
@Override
54-
public void writeTo(StreamOutput out) throws IOException {
67+
public void writeTo(final StreamOutput out) throws IOException {
5568
super.writeTo(out);
56-
out.writeList(operations);
69+
out.writeArray(Translog.Operation::writeOperation, operations);
70+
}
71+
72+
@Override
73+
public boolean equals(final Object o) {
74+
if (this == o) return true;
75+
if (o == null || getClass() != o.getClass()) return false;
76+
final ResyncReplicationRequest that = (ResyncReplicationRequest) o;
77+
return Arrays.equals(operations, that.operations);
78+
}
79+
80+
@Override
81+
public int hashCode() {
82+
return Arrays.hashCode(operations);
5783
}
5884

5985
@Override
@@ -62,7 +88,8 @@ public String toString() {
6288
"shardId=" + shardId +
6389
", timeout=" + timeout +
6490
", index='" + index + '\'' +
65-
", ops=" + operations.size() +
91+
", ops=" + operations.length +
6692
"}";
6793
}
94+
6895
}

‎core/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -688,9 +688,21 @@ public byte[] readByteArray() throws IOException {
688688
return bytes;
689689
}
690690

691-
public <T> T[] readArray(Writeable.Reader<T> reader, IntFunction<T[]> arraySupplier) throws IOException {
692-
int length = readArraySize();
693-
T[] values = arraySupplier.apply(length);
691+
/**
692+
* Reads an array from the stream using the specified {@link org.elasticsearch.common.io.stream.Writeable.Reader} to read array elements
693+
* from the stream. This method can be seen as the reader version of {@link StreamOutput#writeArray(Writeable.Writer, Object[])}. It is
694+
* assumed that the stream first contains a variable-length integer representing the size of the array, and then contains that many
695+
* elements that can be read from the stream.
696+
*
697+
* @param reader the reader used to read individual elements
698+
* @param arraySupplier a supplier used to construct a new array
699+
* @param <T> the type of the elements of the array
700+
* @return an array read from the stream
701+
* @throws IOException if an I/O exception occurs while reading the array
702+
*/
703+
public <T> T[] readArray(final Writeable.Reader<T> reader, final IntFunction<T[]> arraySupplier) throws IOException {
704+
final int length = readArraySize();
705+
final T[] values = arraySupplier.apply(length);
694706
for (int i = 0; i < length; i++) {
695707
values[i] = reader.read(this);
696708
}

‎core/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import java.util.LinkedHashMap;
5959
import java.util.List;
6060
import java.util.Map;
61+
import java.util.function.IntFunction;
6162

6263
/**
6364
* A stream from another node to this node. Technically, it can also be streamed from a byte array but that is mostly for testing.
@@ -706,6 +707,23 @@ public void writeDoubleArray(double[] values) throws IOException {
706707
}
707708
}
708709

710+
/**
711+
* Writes the specified array to the stream using the specified {@link Writer} for each element in the array. This method can be seen as
712+
* writer version of {@link StreamInput#readArray(Writeable.Reader, IntFunction)}. The length of array encoded as a variable-length
713+
* integer is first written to the stream, and then the elements of the array are written to the stream.
714+
*
715+
* @param writer the writer used to write individual elements
716+
* @param array the array
717+
* @param <T> the type of the elements of the array
718+
* @throws IOException if an I/O exception occurs while writing the array
719+
*/
720+
public <T> void writeArray(final Writer<T> writer, final T[] array) throws IOException {
721+
writeVInt(array.length);
722+
for (T value : array) {
723+
writer.write(this, value);
724+
}
725+
}
726+
709727
public <T extends Writeable> void writeArray(T[] array) throws IOException {
710728
writeVInt(array.length);
711729
for (T value: array) {

‎core/src/main/java/org/elasticsearch/index/shard/PrimaryReplicaSyncer.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -218,6 +218,8 @@ public void onFailure(Exception e) {
218218
}
219219
}
220220

221+
private static Translog.Operation[] EMPTY_ARRAY = new Translog.Operation[0];
222+
221223
@Override
222224
protected void doRun() throws Exception {
223225
long size = 0;
@@ -247,7 +249,7 @@ protected void doRun() throws Exception {
247249

248250
if (!operations.isEmpty()) {
249251
task.setPhase("sending_ops");
250-
ResyncReplicationRequest request = new ResyncReplicationRequest(shardId, operations);
252+
ResyncReplicationRequest request = new ResyncReplicationRequest(shardId, operations.toArray(EMPTY_ARRAY));
251253
logger.trace("{} sending batch of [{}][{}] (total sent: [{}], skipped: [{}])", shardId, operations.size(),
252254
new ByteSizeValue(size), totalSentOps.get(), totalSkippedOps.get());
253255
syncAction.sync(request, task, primaryAllocationId, primaryTerm, this);

‎core/src/main/java/org/elasticsearch/index/translog/Translog.java‎

Lines changed: 32 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333
import org.elasticsearch.common.io.stream.ReleasableBytesStreamOutput;
3434
import org.elasticsearch.common.io.stream.StreamInput;
3535
import org.elasticsearch.common.io.stream.StreamOutput;
36-
import org.elasticsearch.common.io.stream.Writeable;
3736
import org.elasticsearch.common.lease.Releasable;
3837
import org.elasticsearch.common.lease.Releasables;
3938
import org.elasticsearch.common.lucene.uid.Versions;
@@ -847,7 +846,7 @@ public interface Snapshot extends Closeable {
847846
* A generic interface representing an operation performed on the transaction log.
848847
* Each is associated with a type.
849848
*/
850-
public interface Operation extends Writeable {
849+
public interface Operation {
851850
enum Type {
852851
@Deprecated
853852
CREATE((byte) 1),
@@ -876,7 +875,7 @@ public static Type fromId(byte id) {
876875
case 4:
877876
return NO_OP;
878877
default:
879-
throw new IllegalArgumentException("No type mapped for [" + id + "]");
878+
throw new IllegalArgumentException("no type mapped for [" + id + "]");
880879
}
881880
}
882881
}
@@ -893,31 +892,44 @@ public static Type fromId(byte id) {
893892

894893
/**
895894
* Reads the type and the operation from the given stream. The operation must be written with
896-
* {@link Operation#writeType(Operation, StreamOutput)}
895+
* {@link Operation#writeOperation(StreamOutput, Operation)}
897896
*/
898-
static Operation readType(StreamInput input) throws IOException {
899-
Translog.Operation.Type type = Translog.Operation.Type.fromId(input.readByte());
897+
static Operation readOperation(final StreamInput input) throws IOException {
898+
final Translog.Operation.Type type = Translog.Operation.Type.fromId(input.readByte());
900899
switch (type) {
901900
case CREATE:
902-
// the deserialization logic in Index was identical to that of Create when create was deprecated
901+
// the de-serialization logic in Index was identical to that of Create when create was deprecated
902+
case INDEX:
903903
return new Index(input);
904904
case DELETE:
905905
return new Delete(input);
906-
case INDEX:
907-
return new Index(input);
908906
case NO_OP:
909907
return new NoOp(input);
910908
default:
911-
throw new IOException("No type for [" + type + "]");
909+
throw new AssertionError("no case for [" + type + "]");
912910
}
913911
}
914912

915913
/**
916914
* Writes the type and translog operation to the given stream
917915
*/
918-
static void writeType(Translog.Operation operation, StreamOutput output) throws IOException {
916+
static void writeOperation(final StreamOutput output, final Operation operation) throws IOException {
919917
output.writeByte(operation.opType().id());
920-
operation.writeTo(output);
918+
switch(operation.opType()) {
919+
case CREATE:
920+
// the serialization logic in Index was identical to that of Create when create was deprecated
921+
case INDEX:
922+
((Index) operation).write(output);
923+
break;
924+
case DELETE:
925+
((Delete) operation).write(output);
926+
break;
927+
case NO_OP:
928+
((NoOp) operation).write(output);
929+
break;
930+
default:
931+
throw new AssertionError("no case for [" + operation.opType() + "]");
932+
}
921933
}
922934

923935
}
@@ -954,7 +966,7 @@ public static class Index implements Operation {
954966
private final String routing;
955967
private final String parent;
956968

957-
public Index(StreamInput in) throws IOException {
969+
private Index(final StreamInput in) throws IOException {
958970
final int format = in.readVInt(); // SERIALIZATION_FORMAT
959971
assert format >= FORMAT_2_X : "format was: " + format;
960972
id = in.readString();
@@ -1067,8 +1079,7 @@ public Source getSource() {
10671079
return new Source(source, routing, parent);
10681080
}
10691081

1070-
@Override
1071-
public void writeTo(StreamOutput out) throws IOException {
1082+
private void write(final StreamOutput out) throws IOException {
10721083
out.writeVInt(SERIALIZATION_FORMAT);
10731084
out.writeString(id);
10741085
out.writeString(type);
@@ -1156,7 +1167,7 @@ public static class Delete implements Operation {
11561167
private final long version;
11571168
private final VersionType versionType;
11581169

1159-
public Delete(StreamInput in) throws IOException {
1170+
private Delete(final StreamInput in) throws IOException {
11601171
final int format = in.readVInt();// SERIALIZATION_FORMAT
11611172
assert format >= FORMAT_5_0 : "format was: " + format;
11621173
if (format >= FORMAT_SINGLE_TYPE) {
@@ -1251,8 +1262,7 @@ public Source getSource() {
12511262
throw new IllegalStateException("trying to read doc source from delete operation");
12521263
}
12531264

1254-
@Override
1255-
public void writeTo(StreamOutput out) throws IOException {
1265+
private void write(final StreamOutput out) throws IOException {
12561266
out.writeVInt(SERIALIZATION_FORMAT);
12571267
out.writeString(type);
12581268
out.writeString(id);
@@ -1322,7 +1332,7 @@ public String reason() {
13221332
return reason;
13231333
}
13241334

1325-
NoOp(final StreamInput in) throws IOException {
1335+
private NoOp(final StreamInput in) throws IOException {
13261336
seqNo = in.readLong();
13271337
primaryTerm = in.readLong();
13281338
reason = in.readString();
@@ -1337,8 +1347,7 @@ public NoOp(final long seqNo, final long primaryTerm, final String reason) {
13371347
this.reason = reason;
13381348
}
13391349

1340-
@Override
1341-
public void writeTo(StreamOutput out) throws IOException {
1350+
private void write(final StreamOutput out) throws IOException {
13421351
out.writeLong(seqNo);
13431352
out.writeLong(primaryTerm);
13441353
out.writeString(reason);
@@ -1440,7 +1449,7 @@ static Translog.Operation readOperation(BufferedChecksumStreamInput in) throws I
14401449
verifyChecksum(in);
14411450
in.reset();
14421451
}
1443-
operation = Translog.Operation.readType(in);
1452+
operation = Translog.Operation.readOperation(in);
14441453
verifyChecksum(in);
14451454
} catch (TranslogCorruptedException e) {
14461455
throw e;
@@ -1483,7 +1492,7 @@ public static void writeOperationNoSize(BufferedChecksumStreamOutput out, Transl
14831492
// because closing it closes the underlying stream, which we don't
14841493
// want to do here.
14851494
out.resetDigest();
1486-
Translog.Operation.writeType(op, out);
1495+
Translog.Operation.writeOperation(out, op);
14871496
long checksum = out.getChecksum();
14881497
out.writeInt((int) checksum);
14891498
}

‎core/src/test/java/org/elasticsearch/action/resync/ResyncReplicationRequestTests.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.elasticsearch.action.resync;
21+
22+
import org.elasticsearch.common.io.stream.BytesStreamOutput;
23+
import org.elasticsearch.common.io.stream.StreamInput;
24+
import org.elasticsearch.common.lucene.uid.Versions;
25+
import org.elasticsearch.index.Index;
26+
import org.elasticsearch.index.VersionType;
27+
import org.elasticsearch.index.shard.ShardId;
28+
import org.elasticsearch.index.translog.Translog;
29+
import org.elasticsearch.test.ESTestCase;
30+
31+
import java.io.IOException;
32+
import java.nio.charset.Charset;
33+
34+
import static org.hamcrest.Matchers.equalTo;
35+
36+
public class ResyncReplicationRequestTests extends ESTestCase {
37+
38+
public void testSerialization() throws IOException {
39+
final byte[] bytes = "{}".getBytes(Charset.forName("UTF-8"));
40+
final Translog.Index index = new Translog.Index("type", "id", 0, Versions.MATCH_ANY, VersionType.INTERNAL, bytes, null, null, -1);
41+
final ShardId shardId = new ShardId(new Index("index", "uuid"), 0);
42+
final ResyncReplicationRequest before = new ResyncReplicationRequest(shardId, new Translog.Operation[]{index});
43+
44+
final BytesStreamOutput out = new BytesStreamOutput();
45+
before.writeTo(out);
46+
47+
final StreamInput in = out.bytes().streamInput();
48+
final ResyncReplicationRequest after = new ResyncReplicationRequest();
49+
after.readFrom(in);
50+
51+
assertThat(after, equalTo(before));
52+
}
53+
54+
}

‎core/src/test/java/org/elasticsearch/index/shard/PrimaryReplicaSyncerTests.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ public void testSyncerSendsOffCorrectDocuments() throws Exception {
4848
AtomicBoolean syncActionCalled = new AtomicBoolean();
4949
PrimaryReplicaSyncer.SyncAction syncAction =
5050
(request, parentTask, allocationId, primaryTerm, listener) -> {
51-
logger.info("Sending off {} operations", request.getOperations().size());
51+
logger.info("Sending off {} operations", request.getOperations().length);
5252
syncActionCalled.set(true);
5353
assertThat(parentTask, instanceOf(PrimaryReplicaSyncer.ResyncTask.class));
5454
listener.onResponse(new ResyncReplicationResponse());
@@ -98,7 +98,7 @@ public void testSyncerOnClosingShard() throws Exception {
9898
CountDownLatch syncCalledLatch = new CountDownLatch(1);
9999
PrimaryReplicaSyncer.SyncAction syncAction =
100100
(request, parentTask, allocationId, primaryTerm, listener) -> {
101-
logger.info("Sending off {} operations", request.getOperations().size());
101+
logger.info("Sending off {} operations", request.getOperations().length);
102102
syncActionCalled.set(true);
103103
syncCalledLatch.countDown();
104104
threadPool.generic().execute(() -> listener.onResponse(new ResyncReplicationResponse()));

0 commit comments

Comments
 (0)