Skip to content

Dry up inputstream to bytesreference #43675

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Jul 2, 2019
Merged

Dry up inputstream to bytesreference #43675

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d07e40b
Dry up Reading InputStream to BytesReference
original-brownbear Jun 27, 2019
6464f34
fix import
original-brownbear Jun 27, 2019
083ca36
Merge remote-tracking branch 'elastic/master' into dry-up-inputstream…
original-brownbear Jul 2, 2019
caace70
CR: remove senseless copying
original-brownbear Jul 2, 2019
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
10 changes: 2 additions & 8 deletions server/src/main/java/org/elasticsearch/common/compress/CompressorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,9 @@

import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.internal.io.Streams;

import java.io.IOException;
import java.util.Objects;
Expand Down Expand Up @@ -93,10 +91,6 @@ public static BytesReference uncompress(BytesReference bytes) throws IOException
}

private static BytesReference uncompress(BytesReference bytes, Compressor compressor) throws IOException {
StreamInput compressed = compressor.streamInput(bytes.streamInput());
BytesStreamOutput bStream = new BytesStreamOutput();
Streams.copy(compressed, bStream);
compressed.close();
return bStream.bytes();
return Streams.readFully(compressor.streamInput(bytes.streamInput()));
}
}
12 changes: 12 additions & 0 deletions server/src/main/java/org/elasticsearch/common/io/Streams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.io.stream.BytesStream;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.StreamOutput;

import java.io.BufferedReader;
Expand Down Expand Up @@ -226,6 +227,17 @@ public static BytesStream flushOnCloseStream(BytesStream os) {
return new FlushOnCloseOutputStream(os);
}

/**
* Reads all bytes from the given {@link InputStream} and closes it afterwards.
*/
public static BytesReference readFully(InputStream in) throws IOException {
try (InputStream inputStream = in) {
BytesStreamOutput out = new BytesStreamOutput();
copy(inputStream, out);
return out.bytes();
}
}

/**
* A wrapper around a {@link BytesStream} that makes the close operation a flush. This is
* needed as sometimes a stream will be closed but the bytes that the stream holds still need
Expand Down
40 changes: 13 additions & 27 deletions server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.compress.NotXContentException;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.lucene.Lucene;
import org.elasticsearch.common.lucene.store.InputStreamIndexInput;
Expand All @@ -61,10 +62,8 @@
import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.internal.io.Streams;
import org.elasticsearch.index.mapper.MapperService;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.snapshots.IndexShardRestoreFailedException;
Expand Down Expand Up @@ -645,32 +644,23 @@ public RepositoryData getRepositoryData() {
final String snapshotsIndexBlobName = INDEX_FILE_PREFIX + Long.toString(indexGen);

RepositoryData repositoryData;
try (InputStream blob = blobContainer().readBlob(snapshotsIndexBlobName)) {
BytesStreamOutput out = new BytesStreamOutput();
Streams.copy(blob, out);
// EMPTY is safe here because RepositoryData#fromXContent calls namedObject
try (XContentParser parser = XContentHelper.createParser(NamedXContentRegistry.EMPTY,
LoggingDeprecationHandler.INSTANCE, out.bytes(), XContentType.JSON)) {
repositoryData = RepositoryData.snapshotsFromXContent(parser, indexGen);
} catch (NotXContentException e) {
logger.warn("[{}] index blob is not valid x-content [{} bytes]", snapshotsIndexBlobName, out.bytes().length());
throw e;
}
// EMPTY is safe here because RepositoryData#fromXContent calls namedObject
try (InputStream blob = blobContainer().readBlob(snapshotsIndexBlobName);
XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
LoggingDeprecationHandler.INSTANCE, blob)) {
repositoryData = RepositoryData.snapshotsFromXContent(parser, indexGen);
}

// now load the incompatible snapshot ids, if they exist
try (InputStream blob = blobContainer().readBlob(INCOMPATIBLE_SNAPSHOTS_BLOB)) {
BytesStreamOutput out = new BytesStreamOutput();
Streams.copy(blob, out);
try (XContentParser parser = XContentHelper.createParser(NamedXContentRegistry.EMPTY,
LoggingDeprecationHandler.INSTANCE, out.bytes(), XContentType.JSON)) {
repositoryData = repositoryData.incompatibleSnapshotsFromXContent(parser);
}
try (InputStream blob = blobContainer().readBlob(INCOMPATIBLE_SNAPSHOTS_BLOB);
XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
LoggingDeprecationHandler.INSTANCE, blob)) {
repositoryData = repositoryData.incompatibleSnapshotsFromXContent(parser);
} catch (NoSuchFileException e) {
if (isReadOnly()) {
logger.debug("[{}] Incompatible snapshots blob [{}] does not exist, the likely " +
"reason is that there are no incompatible snapshots in the repository",
metadata.name(), INCOMPATIBLE_SNAPSHOTS_BLOB);
"reason is that there are no incompatible snapshots in the repository",
metadata.name(), INCOMPATIBLE_SNAPSHOTS_BLOB);
} else {
// write an empty incompatible-snapshots blob - we do this so that there
// is a blob present, which helps speed up some cloud-based repositories
Expand Down Expand Up @@ -778,11 +768,7 @@ long latestIndexBlobId() throws IOException {

// package private for testing
long readSnapshotIndexLatestBlob() throws IOException {
try (InputStream blob = blobContainer().readBlob(INDEX_LATEST_BLOB)) {
BytesStreamOutput out = new BytesStreamOutput();
Streams.copy(blob, out);
return Numbers.bytesToLong(out.bytes().toBytesRef());
}
return Numbers.bytesToLong(Streams.readFully(blobContainer().readBlob(INDEX_LATEST_BLOB)).toBytesRef());
Copy link
Member

Choose a reason for hiding this comment

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

We expect this file to be very small so using Streams.readFully()can makes sense here.

}

private long listBlobsToGetLatestIndexId() throws IOException {
Expand Down
33 changes: 15 additions & 18 deletions server/src/main/java/org/elasticsearch/repositories/blobstore/ChecksumBlobStoreFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.compress.CompressorFactory;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.lucene.store.ByteArrayIndexInput;
Expand All @@ -42,7 +43,6 @@
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.internal.io.Streams;
import org.elasticsearch.gateway.CorruptStateException;
import org.elasticsearch.snapshots.SnapshotInfo;

Expand Down Expand Up @@ -149,24 +149,21 @@ public String blobName(String name) {
* @param blobName blob name
*/
public T readBlob(BlobContainer blobContainer, String blobName) throws IOException {
try (InputStream inputStream = blobContainer.readBlob(blobName)) {
ByteArrayOutputStream out = new ByteArrayOutputStream();
Streams.copy(inputStream, out);
final byte[] bytes = out.toByteArray();
final String resourceDesc = "ChecksumBlobStoreFormat.readBlob(blob=\"" + blobName + "\")";
try (ByteArrayIndexInput indexInput = new ByteArrayIndexInput(resourceDesc, bytes)) {
CodecUtil.checksumEntireFile(indexInput);
CodecUtil.checkHeader(indexInput, codec, VERSION, VERSION);
long filePointer = indexInput.getFilePointer();
long contentSize = indexInput.length() - CodecUtil.footerLength() - filePointer;
try (XContentParser parser = XContentHelper.createParser(namedXContentRegistry, LoggingDeprecationHandler.INSTANCE,
new BytesArray(bytes, (int) filePointer, (int) contentSize))) {
return reader.apply(parser);
}
} catch (CorruptIndexException | IndexFormatTooOldException | IndexFormatTooNewException ex) {
// we trick this into a dedicated exception with the original stacktrace
throw new CorruptStateException(ex);
final BytesReference bytes = Streams.readFully(blobContainer.readBlob(blobName));
final String resourceDesc = "ChecksumBlobStoreFormat.readBlob(blob=\"" + blobName + "\")";
try (ByteArrayIndexInput indexInput =
new ByteArrayIndexInput(resourceDesc, BytesReference.toBytes(bytes))) {
CodecUtil.checksumEntireFile(indexInput);
CodecUtil.checkHeader(indexInput, codec, VERSION, VERSION);
long filePointer = indexInput.getFilePointer();
long contentSize = indexInput.length() - CodecUtil.footerLength() - filePointer;
try (XContentParser parser = XContentHelper.createParser(namedXContentRegistry, LoggingDeprecationHandler.INSTANCE,
bytes.slice((int) filePointer, (int) contentSize))) {
return reader.apply(parser);
}
} catch (CorruptIndexException | IndexFormatTooOldException | IndexFormatTooNewException ex) {
// we trick this into a dedicated exception with the original stacktrace
throw new CorruptStateException(ex);
}
}

Expand Down
15 changes: 4 additions & 11 deletions test/framework/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreTestUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,11 @@
import org.elasticsearch.action.support.PlainActionFuture;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.blobstore.BlobContainer;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.internal.io.Streams;
import org.elasticsearch.repositories.IndexId;
import org.elasticsearch.repositories.RepositoriesService;
import org.elasticsearch.repositories.RepositoryData;
Expand Down Expand Up @@ -81,14 +78,10 @@ protected void doRun() throws Exception {
}
assertIndexGenerations(blobContainer, latestGen);
final RepositoryData repositoryData;
try (InputStream inputStream = blobContainer.readBlob("index-" + latestGen);
BytesStreamOutput out = new BytesStreamOutput()) {
Streams.copy(inputStream, out);
try (XContentParser parser =
XContentHelper.createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE,
out.bytes(), XContentType.JSON)) {
repositoryData = RepositoryData.snapshotsFromXContent(parser, latestGen);
}
try (InputStream blob = blobContainer.readBlob("index-" + latestGen);
XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
LoggingDeprecationHandler.INSTANCE, blob)) {
repositoryData = RepositoryData.snapshotsFromXContent(parser, latestGen);
}
assertIndexUUIDs(blobContainer, repositoryData);
assertSnapshotUUIDs(blobContainer, repositoryData);
Expand Down
11 changes: 2 additions & 9 deletions x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/template/TemplateUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,17 +16,15 @@
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.compress.CompressedXContent;
import org.elasticsearch.common.compress.NotXContentException;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.internal.io.Streams;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.function.Predicate;
import java.util.regex.Pattern;
Expand Down Expand Up @@ -73,12 +71,7 @@ public static String loadTemplate(String resource, String version, String versio
* Loads a resource from the classpath and returns it as a {@link BytesReference}
*/
public static BytesReference load(String name) throws IOException {
try (InputStream is = TemplateUtils.class.getResourceAsStream(name)) {
try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
Streams.copy(is, out);
return new BytesArray(out.toByteArray());
}
}
return Streams.readFully(TemplateUtils.class.getResourceAsStream(name));
}

/**
Expand Down
8 changes: 2 additions & 6 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,15 +11,14 @@
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentElasticsearchExtension;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.internal.io.Streams;
import org.elasticsearch.index.mapper.DateFieldMapper;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.xpack.core.ml.action.FlushJobAction;
Expand All @@ -38,7 +37,6 @@
import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
import org.elasticsearch.xpack.ml.notifications.Auditor;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
Expand Down Expand Up @@ -407,9 +405,7 @@ private DataCounts postData(InputStream inputStream, XContentType xContentType)
throws IOException {
PostDataAction.Request request = new PostDataAction.Request(jobId);
request.setDataDescription(dataDescription);
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
Streams.copy(inputStream, outputStream);
request.setContent(new BytesArray(outputStream.toByteArray()), xContentType);
request.setContent(Streams.readFully(inputStream), xContentType);
try (ThreadContext.StoredContext ignore = client.threadPool().getThreadContext().stashWithOrigin(ML_ORIGIN)) {
PostDataAction.Response response = client.execute(PostDataAction.INSTANCE, request).actionGet();
return response.getDataCounts();
Expand Down
13 changes: 2 additions & 11 deletions ...nitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/ClusterAlertsUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,11 @@

import org.elasticsearch.Version;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.io.Streams;
import org.elasticsearch.common.settings.SettingsException;
import org.elasticsearch.core.internal.io.Streams;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.List;
import java.util.Locale;
Expand Down Expand Up @@ -124,13 +121,7 @@ public static String loadWatch(final ClusterService clusterService, final String
}

private static BytesReference loadResource(final String resource) throws IOException {
try (InputStream is = ClusterAlertsUtil.class.getResourceAsStream(resource)) {
try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
Streams.copy(is, out);

return new BytesArray(out.toByteArray());
}
}
return Streams.readFully(ClusterAlertsUtil.class.getResourceAsStream(resource));
}

/**
Expand Down