Skip to content

Commit 2b5c0d7

Browse files
dnhatndnhatn
committed
Merge branch 'master' into ccr
* master: Removing erroneous repeat Adapt bwc versions after backporting #30983 to 6.4 [Tests] Muting RatedRequestsTests#testXContentParsingIsNotLenient TEST: Retry synced-flush if ongoing ops on primary (#30978) Fix docs build. Only auto-update license signature if all nodes ready (#30859) Add BlobContainer.writeBlobAtomic() (#30902) Add a doc value format to binary fields. (#30860)
2 parents 530089f + 14c4088 commit 2b5c0d7

File tree

40 files changed

+406
-216
lines changed

40 files changed

+406
-216
lines changed

‎buildSrc/src/main/resources/checkstyle_suppressions.xml

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -505,8 +505,6 @@
505505
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]cluster[/\\]settings[/\\]ClusterSettingsIT.java" checks="LineLength" />
506506
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]cluster[/\\]shards[/\\]ClusterSearchShardsIT.java" checks="LineLength" />
507507
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]cluster[/\\]structure[/\\]RoutingIteratorTests.java" checks="LineLength" />
508-
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]blobstore[/\\]FsBlobStoreContainerTests.java" checks="LineLength" />
509-
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]blobstore[/\\]FsBlobStoreTests.java" checks="LineLength" />
510508
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]breaker[/\\]MemoryCircuitBreakerTests.java" checks="LineLength" />
511509
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]geo[/\\]ShapeBuilderTests.java" checks="LineLength" />
512510
<suppress files="server[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]common[/\\]hash[/\\]MessageDigestsTests.java" checks="LineLength" />

‎docs/reference/query-dsl/query-string-syntax.asciidoc

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -254,7 +254,6 @@ would look like this:
254254
}
255255
}
256256

257-
****
258257

259258
===== Grouping
260259

‎modules/rank-eval/src/test/java/org/elasticsearch/index/rankeval/RatedRequestsTests.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,7 @@ public void testXContentRoundtrip() throws IOException {
131131
}
132132
}
133133

134+
@AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/31104")
134135
public void testXContentParsingIsNotLenient() throws IOException {
135136
RatedRequest testItem = createTestItem(randomBoolean());
136137
XContentType xContentType = randomFrom(XContentType.values());

‎server/src/main/java/org/elasticsearch/action/admin/cluster/reroute/ClusterRerouteResponse.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ public RoutingExplanations getExplanations() {
6363

6464
@Override
6565
public void readFrom(StreamInput in) throws IOException {
66-
if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
66+
if (in.getVersion().onOrAfter(Version.V_6_4_0)) {
6767
super.readFrom(in);
6868
state = ClusterState.readFrom(in, null);
6969
explanations = RoutingExplanations.readFrom(in);
@@ -76,7 +76,7 @@ public void readFrom(StreamInput in) throws IOException {
7676

7777
@Override
7878
public void writeTo(StreamOutput out) throws IOException {
79-
if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
79+
if (out.getVersion().onOrAfter(Version.V_6_4_0)) {
8080
super.writeTo(out);
8181
state.writeTo(out);
8282
RoutingExplanations.writeTo(explanations, out);

‎server/src/main/java/org/elasticsearch/action/admin/cluster/settings/ClusterUpdateSettingsResponse.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ public class ClusterUpdateSettingsResponse extends AcknowledgedResponse {
6868

6969
@Override
7070
public void readFrom(StreamInput in) throws IOException {
71-
if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
71+
if (in.getVersion().onOrAfter(Version.V_6_4_0)) {
7272
super.readFrom(in);
7373
transientSettings = Settings.readSettingsFromStream(in);
7474
persistentSettings = Settings.readSettingsFromStream(in);
@@ -89,7 +89,7 @@ public Settings getPersistentSettings() {
8989

9090
@Override
9191
public void writeTo(StreamOutput out) throws IOException {
92-
if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
92+
if (out.getVersion().onOrAfter(Version.V_6_4_0)) {
9393
super.writeTo(out);
9494
Settings.writeSettingsToStream(transientSettings, out);
9595
Settings.writeSettingsToStream(persistentSettings, out);

‎server/src/main/java/org/elasticsearch/action/admin/indices/rollover/RolloverResponse.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ public boolean isRolledOver() {
115115

116116
@Override
117117
public void readFrom(StreamInput in) throws IOException {
118-
if (in.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
118+
if (in.getVersion().onOrAfter(Version.V_6_4_0)) {
119119
super.readFrom(in);
120120
oldIndex = in.readString();
121121
newIndex = in.readString();
@@ -144,7 +144,7 @@ public void readFrom(StreamInput in) throws IOException {
144144

145145
@Override
146146
public void writeTo(StreamOutput out) throws IOException {
147-
if (out.getVersion().onOrAfter(Version.V_7_0_0_alpha1)) {
147+
if (out.getVersion().onOrAfter(Version.V_6_4_0)) {
148148
super.writeTo(out);
149149
out.writeString(oldIndex);
150150
out.writeString(newIndex);

‎server/src/main/java/org/elasticsearch/common/blobstore/BlobContainer.java

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,29 @@ public interface BlobContainer {
7474
*/
7575
void writeBlob(String blobName, InputStream inputStream, long blobSize) throws IOException;
7676

77+
/**
78+
* Reads blob content from the input stream and writes it to the container in a new blob with the given name,
79+
* using an atomic write operation if the implementation supports it. When the BlobContainer implementation
80+
* does not provide a specific implementation of writeBlobAtomic(String, InputStream, long), then
81+
* the {@link #writeBlob(String, InputStream, long)} method is used.
82+
*
83+
* This method assumes the container does not already contain a blob of the same blobName. If a blob by the
84+
* same name already exists, the operation will fail and an {@link IOException} will be thrown.
85+
*
86+
* @param blobName
87+
* The name of the blob to write the contents of the input stream to.
88+
* @param inputStream
89+
* The input stream from which to retrieve the bytes to write to the blob.
90+
* @param blobSize
91+
* The size of the blob to be written, in bytes. It is implementation dependent whether
92+
* this value is used in writing the blob to the repository.
93+
* @throws FileAlreadyExistsException if a blob by the same name already exists
94+
* @throws IOException if the input stream could not be read, or the target blob could not be written to.
95+
*/
96+
default void writeBlobAtomic(final String blobName, final InputStream inputStream, final long blobSize) throws IOException {
97+
writeBlob(blobName, inputStream, blobSize);
98+
}
99+
77100
/**
78101
* Deletes a blob with giving name, if the blob exists. If the blob does not exist,
79102
* this method throws a NoSuchFileException.

‎server/src/main/java/org/elasticsearch/common/blobstore/fs/FsBlobContainer.java

Lines changed: 46 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,12 @@
1919

2020
package org.elasticsearch.common.blobstore.fs;
2121

22-
import org.elasticsearch.core.internal.io.IOUtils;
22+
import org.elasticsearch.common.UUIDs;
2323
import org.elasticsearch.common.blobstore.BlobMetaData;
2424
import org.elasticsearch.common.blobstore.BlobPath;
2525
import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
2626
import org.elasticsearch.common.blobstore.support.PlainBlobMetaData;
27+
import org.elasticsearch.core.internal.io.IOUtils;
2728
import org.elasticsearch.core.internal.io.Streams;
2829

2930
import java.io.BufferedInputStream;
@@ -56,8 +57,9 @@
5657
*/
5758
public class FsBlobContainer extends AbstractBlobContainer {
5859

59-
protected final FsBlobStore blobStore;
60+
private static final String TEMP_FILE_PREFIX = "pending-";
6061

62+
protected final FsBlobStore blobStore;
6163
protected final Path path;
6264

6365
public FsBlobContainer(FsBlobStore blobStore, BlobPath blobPath, Path path) {
@@ -131,6 +133,48 @@ public void writeBlob(String blobName, InputStream inputStream, long blobSize) t
131133
IOUtils.fsync(path, true);
132134
}
133135

136+
@Override
137+
public void writeBlobAtomic(final String blobName, final InputStream inputStream, final long blobSize) throws IOException {
138+
final String tempBlob = tempBlobName(blobName);
139+
final Path tempBlobPath = path.resolve(tempBlob);
140+
try {
141+
try (OutputStream outputStream = Files.newOutputStream(tempBlobPath, StandardOpenOption.CREATE_NEW)) {
142+
Streams.copy(inputStream, outputStream);
143+
}
144+
IOUtils.fsync(tempBlobPath, false);
145+
146+
final Path blobPath = path.resolve(blobName);
147+
// If the target file exists then Files.move() behaviour is implementation specific
148+
// the existing file might be replaced or this method fails by throwing an IOException.
149+
if (Files.exists(blobPath)) {
150+
throw new FileAlreadyExistsException("blob [" + blobPath + "] already exists, cannot overwrite");
151+
}
152+
Files.move(tempBlobPath, blobPath, StandardCopyOption.ATOMIC_MOVE);
153+
} catch (IOException ex) {
154+
try {
155+
deleteBlobIgnoringIfNotExists(tempBlob);
156+
} catch (IOException e) {
157+
ex.addSuppressed(e);
158+
}
159+
throw ex;
160+
} finally {
161+
IOUtils.fsync(path, true);
162+
}
163+
}
164+
165+
public static String tempBlobName(final String blobName) {
166+
return "pending-" + blobName + "-" + UUIDs.randomBase64UUID();
167+
}
168+
169+
/**
170+
* Returns true if the blob is a leftover temporary blob.
171+
*
172+
* The temporary blobs might be left after failed atomic write operation.
173+
*/
174+
public static boolean isTempBlobName(final String blobName) {
175+
return blobName.startsWith(TEMP_FILE_PREFIX);
176+
}
177+
134178
@Override
135179
public void move(String source, String target) throws IOException {
136180
Path sourcePath = path.resolve(source);

‎server/src/main/java/org/elasticsearch/index/mapper/BinaryFieldMapper.java

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@
4040
import org.elasticsearch.index.fielddata.plain.BytesBinaryDVIndexFieldData;
4141
import org.elasticsearch.index.query.QueryShardContext;
4242
import org.elasticsearch.index.query.QueryShardException;
43+
import org.elasticsearch.search.DocValueFormat;
44+
import org.joda.time.DateTimeZone;
4345

4446
import java.io.IOException;
4547
import java.util.Base64;
@@ -104,6 +106,10 @@ public String typeName() {
104106
return CONTENT_TYPE;
105107
}
106108

109+
@Override
110+
public DocValueFormat docValueFormat(String format, DateTimeZone timeZone) {
111+
return DocValueFormat.BINARY;
112+
}
107113

108114
@Override
109115
public BytesReference valueForDisplay(Object value) {

‎server/src/main/java/org/elasticsearch/indices/flush/SyncedFlushService.java

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -507,18 +507,7 @@ private InFlightOpsResponse performInFlightOps(InFlightOpsRequest request) {
507507
if (indexShard.routingEntry().primary() == false) {
508508
throw new IllegalStateException("[" + request.shardId() +"] expected a primary shard");
509509
}
510-
if (Assertions.ENABLED) {
511-
if (logger.isTraceEnabled()) {
512-
logger.trace("in flight operations {}, acquirers {}", indexShard.getActiveOperationsCount(), indexShard.getActiveOperations());
513-
}
514-
}
515510
int opCount = indexShard.getActiveOperationsCount();
516-
// Need to snapshot the debug info twice as it's updated concurrently with the permit count.
517-
if (Assertions.ENABLED) {
518-
if (logger.isTraceEnabled()) {
519-
logger.trace("in flight operations {}, acquirers {}", indexShard.getActiveOperationsCount(), indexShard.getActiveOperations());
520-
}
521-
}
522511
return new InFlightOpsResponse(opCount);
523512
}
524513

‎server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

Lines changed: 4 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
import org.elasticsearch.common.blobstore.BlobMetaData;
5151
import org.elasticsearch.common.blobstore.BlobPath;
5252
import org.elasticsearch.common.blobstore.BlobStore;
53+
import org.elasticsearch.common.blobstore.fs.FsBlobContainer;
5354
import org.elasticsearch.common.bytes.BytesArray;
5455
import org.elasticsearch.common.bytes.BytesReference;
5556
import org.elasticsearch.common.collect.Tuple;
@@ -555,10 +556,8 @@ public String startVerification() {
555556
String blobName = "master.dat";
556557
BytesArray bytes = new BytesArray(testBytes);
557558
try (InputStream stream = bytes.streamInput()) {
558-
testContainer.writeBlob(blobName + "-temp", stream, bytes.length());
559+
testContainer.writeBlobAtomic(blobName, stream, bytes.length());
559560
}
560-
// Make sure that move is supported
561-
testContainer.move(blobName + "-temp", blobName);
562561
return seed;
563562
}
564563
} catch (IOException exp) {
@@ -774,18 +773,8 @@ private long listBlobsToGetLatestIndexId() throws IOException {
774773
}
775774

776775
private void writeAtomic(final String blobName, final BytesReference bytesRef) throws IOException {
777-
final String tempBlobName = "pending-" + blobName + "-" + UUIDs.randomBase64UUID();
778776
try (InputStream stream = bytesRef.streamInput()) {
779-
snapshotsBlobContainer.writeBlob(tempBlobName, stream, bytesRef.length());
780-
snapshotsBlobContainer.move(tempBlobName, blobName);
781-
} catch (IOException ex) {
782-
// temporary blob creation or move failed - try cleaning up
783-
try {
784-
snapshotsBlobContainer.deleteBlobIgnoringIfNotExists(tempBlobName);
785-
} catch (IOException e) {
786-
ex.addSuppressed(e);
787-
}
788-
throw ex;
777+
snapshotsBlobContainer.writeBlobAtomic(blobName, stream, bytesRef.length());
789778
}
790779
}
791780

@@ -955,7 +944,7 @@ protected void finalize(final List<SnapshotFiles> snapshots,
955944
// Delete temporary index files first, as we might otherwise fail in the next step creating the new index file if an earlier
956945
// attempt to write an index file with this generation failed mid-way after creating the temporary file.
957946
for (final String blobName : blobs.keySet()) {
958-
if (indexShardSnapshotsFormat.isTempBlobName(blobName)) {
947+
if (FsBlobContainer.isTempBlobName(blobName)) {
959948
try {
960949
blobContainer.deleteBlobIgnoringIfNotExists(blobName);
961950
} catch (IOException e) {

‎server/src/main/java/org/elasticsearch/repositories/blobstore/ChecksumBlobStoreFormat.java

Lines changed: 20 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.lucene.index.IndexFormatTooNewException;
2424
import org.apache.lucene.index.IndexFormatTooOldException;
2525
import org.apache.lucene.store.OutputStreamIndexOutput;
26+
import org.elasticsearch.common.CheckedConsumer;
2627
import org.elasticsearch.common.CheckedFunction;
2728
import org.elasticsearch.common.blobstore.BlobContainer;
2829
import org.elasticsearch.common.bytes.BytesArray;
@@ -52,8 +53,6 @@
5253
*/
5354
public class ChecksumBlobStoreFormat<T extends ToXContent> extends BlobStoreFormat<T> {
5455

55-
private static final String TEMP_FILE_PREFIX = "pending-";
56-
5756
private static final XContentType DEFAULT_X_CONTENT_TYPE = XContentType.SMILE;
5857

5958
// The format version
@@ -120,7 +119,7 @@ public T readBlob(BlobContainer blobContainer, String blobName) throws IOExcepti
120119
}
121120

122121
/**
123-
* Writes blob in atomic manner with resolving the blob name using {@link #blobName} and {@link #tempBlobName} methods.
122+
* Writes blob in atomic manner with resolving the blob name using {@link #blobName} method.
124123
* <p>
125124
* The blob will be compressed and checksum will be written if required.
126125
*
@@ -131,20 +130,12 @@ public T readBlob(BlobContainer blobContainer, String blobName) throws IOExcepti
131130
* @param name blob name
132131
*/
133132
public void writeAtomic(T obj, BlobContainer blobContainer, String name) throws IOException {
134-
String blobName = blobName(name);
135-
String tempBlobName = tempBlobName(name);
136-
writeBlob(obj, blobContainer, tempBlobName);
137-
try {
138-
blobContainer.move(tempBlobName, blobName);
139-
} catch (IOException ex) {
140-
// Move failed - try cleaning up
141-
try {
142-
blobContainer.deleteBlob(tempBlobName);
143-
} catch (Exception e) {
144-
ex.addSuppressed(e);
133+
final String blobName = blobName(name);
134+
writeTo(obj, blobName, bytesArray -> {
135+
try (InputStream stream = bytesArray.streamInput()) {
136+
blobContainer.writeBlobAtomic(blobName, stream, bytesArray.length());
145137
}
146-
throw ex;
147-
}
138+
});
148139
}
149140

150141
/**
@@ -157,51 +148,35 @@ public void writeAtomic(T obj, BlobContainer blobContainer, String name) throws
157148
* @param name blob name
158149
*/
159150
public void write(T obj, BlobContainer blobContainer, String name) throws IOException {
160-
String blobName = blobName(name);
161-
writeBlob(obj, blobContainer, blobName);
151+
final String blobName = blobName(name);
152+
writeTo(obj, blobName, bytesArray -> {
153+
try (InputStream stream = bytesArray.streamInput()) {
154+
blobContainer.writeBlob(blobName, stream, bytesArray.length());
155+
}
156+
});
162157
}
163158

164-
/**
165-
* Writes blob in atomic manner without resolving the blobName using using {@link #blobName} method.
166-
* <p>
167-
* The blob will be compressed and checksum will be written if required.
168-
*
169-
* @param obj object to be serialized
170-
* @param blobContainer blob container
171-
* @param blobName blob name
172-
*/
173-
protected void writeBlob(T obj, BlobContainer blobContainer, String blobName) throws IOException {
174-
BytesReference bytes = write(obj);
175-
try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
159+
private void writeTo(final T obj, final String blobName, final CheckedConsumer<BytesArray, IOException> consumer) throws IOException {
160+
final BytesReference bytes = write(obj);
161+
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
176162
final String resourceDesc = "ChecksumBlobStoreFormat.writeBlob(blob=\"" + blobName + "\")";
177-
try (OutputStreamIndexOutput indexOutput = new OutputStreamIndexOutput(resourceDesc, blobName, byteArrayOutputStream, BUFFER_SIZE)) {
163+
try (OutputStreamIndexOutput indexOutput = new OutputStreamIndexOutput(resourceDesc, blobName, outputStream, BUFFER_SIZE)) {
178164
CodecUtil.writeHeader(indexOutput, codec, VERSION);
179165
try (OutputStream indexOutputOutputStream = new IndexOutputOutputStream(indexOutput) {
180166
@Override
181167
public void close() throws IOException {
182168
// this is important since some of the XContentBuilders write bytes on close.
183169
// in order to write the footer we need to prevent closing the actual index input.
184-
} }) {
170+
}
171+
}) {
185172
bytes.writeTo(indexOutputOutputStream);
186173
}
187174
CodecUtil.writeFooter(indexOutput);
188175
}
189-
BytesArray bytesArray = new BytesArray(byteArrayOutputStream.toByteArray());
190-
try (InputStream stream = bytesArray.streamInput()) {
191-
blobContainer.writeBlob(blobName, stream, bytesArray.length());
192-
}
176+
consumer.accept(new BytesArray(outputStream.toByteArray()));
193177
}
194178
}
195179

196-
/**
197-
* Returns true if the blob is a leftover temporary blob.
198-
*
199-
* The temporary blobs might be left after failed atomic write operation.
200-
*/
201-
public boolean isTempBlobName(String blobName) {
202-
return blobName.startsWith(ChecksumBlobStoreFormat.TEMP_FILE_PREFIX);
203-
}
204-
205180
protected BytesReference write(T obj) throws IOException {
206181
try (BytesStreamOutput bytesStreamOutput = new BytesStreamOutput()) {
207182
if (compress) {
@@ -222,10 +197,4 @@ protected void write(T obj, StreamOutput streamOutput) throws IOException {
222197
builder.endObject();
223198
}
224199
}
225-
226-
227-
protected String tempBlobName(String name) {
228-
return TEMP_FILE_PREFIX + String.format(Locale.ROOT, blobNameFormat, name);
229-
}
230-
231200
}

0 commit comments

Comments
 (0)