You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ae...@apache.org on 2016/05/13 18:01:07 UTC
[30/41] hadoop git commit: HADOOP-13028 add low level counter metrics
for S3A;
use in read performance tests. contributed by: stevel patch includes
HADOOP-12844 Recover when S3A fails on IOException in read() HADOOP-13058 S3A
FS fails during init against
HADOOP-13028 add low level counter metrics for S3A; use in read performance tests. contributed by: stevel
patch includes
HADOOP-12844 Recover when S3A fails on IOException in read()
HADOOP-13058 S3A FS fails during init against a read-only FS if multipart purge
HADOOP-13047 S3a Forward seek in stream length to be configurable
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/27c4e90e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/27c4e90e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/27c4e90e
Branch: refs/heads/HDFS-1312
Commit: 27c4e90efce04e1b1302f668b5eb22412e00d033
Parents: 6b53802
Author: Steve Loughran <st...@apache.org>
Authored: Thu May 12 19:24:20 2016 +0100
Committer: Steve Loughran <st...@apache.org>
Committed: Thu May 12 19:24:20 2016 +0100
----------------------------------------------------------------------
.../org/apache/hadoop/fs/FSDataInputStream.java | 9 +
.../hadoop/metrics2/MetricStringBuilder.java | 141 ++++++
.../hadoop/metrics2/lib/MutableCounterLong.java | 2 +-
.../src/main/resources/core-default.xml | 10 +-
.../hadoop-aws/dev-support/findbugs-exclude.xml | 357 +--------------
.../apache/hadoop/fs/s3/FileSystemStore.java | 4 +-
.../org/apache/hadoop/fs/s3/S3Credentials.java | 2 +
.../fs/s3a/AnonymousAWSCredentialsProvider.java | 4 +
.../fs/s3a/BasicAWSCredentialsProvider.java | 4 +
.../org/apache/hadoop/fs/s3a/Constants.java | 18 +-
.../hadoop/fs/s3a/S3AFastOutputStream.java | 11 +-
.../org/apache/hadoop/fs/s3a/S3AFileStatus.java | 12 +-
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 459 +++++++++++--------
.../apache/hadoop/fs/s3a/S3AInputStream.java | 342 ++++++++++----
.../hadoop/fs/s3a/S3AInstrumentation.java | 457 ++++++++++++++++++
.../apache/hadoop/fs/s3a/S3AOutputStream.java | 40 +-
.../src/site/markdown/tools/hadoop-aws/index.md | 68 ++-
.../hadoop/fs/s3a/scale/S3AScaleTestBase.java | 191 +++++++-
.../fs/s3a/scale/TestS3ADeleteManyFiles.java | 1 -
.../scale/TestS3AInputStreamPerformance.java | 285 ++++++++++++
.../src/test/resources/log4j.properties | 3 +
21 files changed, 1753 insertions(+), 667 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java
index da98769..640db59 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataInputStream.java
@@ -234,4 +234,13 @@ public class FSDataInputStream extends DataInputStream
"support unbuffering.");
}
}
+
+ /**
+ * String value. Includes the string value of the inner stream
+ * @return the stream
+ */
+ @Override
+ public String toString() {
+ return super.toString() + ": " + in;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/MetricStringBuilder.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/MetricStringBuilder.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/MetricStringBuilder.java
new file mode 100644
index 0000000..18a499a
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/MetricStringBuilder.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.metrics2;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Build a string dump of the metrics.
+ *
+ * The {@link #toString()} operator dumps out all values collected.
+ *
+ * Every entry is formatted as
+ * {@code prefix + name + separator + value + suffix}
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class MetricStringBuilder extends MetricsRecordBuilder {
+
+ private final StringBuilder builder = new StringBuilder(256);
+
+ private final String prefix;
+ private final String suffix;
+ private final String separator;
+ private final MetricsCollector parent;
+
+ /**
+ * Build an instance.
+ * @param parent parent collector. Unused in this instance; only used for
+ * the {@link #parent()} method
+ * @param prefix string before each entry
+ * @param separator separator between name and value
+ * @param suffix suffix after each entry
+ */
+ public MetricStringBuilder(MetricsCollector parent,
+ String prefix,
+ String separator,
+ String suffix) {
+ this.parent = parent;
+ this.prefix = prefix;
+ this.suffix = suffix;
+ this.separator = separator;
+ }
+
+ public MetricStringBuilder add(MetricsInfo info, Object value) {
+ return tuple(info.name(), value.toString());
+ }
+
+ /**
+ * Add any key,val pair to the string, between the prefix and suffix,
+ * separated by the separator.
+ * @param key key
+ * @param value value
+ * @return this instance
+ */
+ public MetricStringBuilder tuple(String key, String value) {
+ builder.append(prefix)
+ .append(key)
+ .append(separator)
+ .append(value)
+ .append(suffix);
+ return this;
+ }
+
+ @Override
+ public MetricsRecordBuilder tag(MetricsInfo info, String value) {
+ return add(info, value);
+ }
+
+ @Override
+ public MetricsRecordBuilder add(MetricsTag tag) {
+ return tuple(tag.name(), tag.value());
+ }
+
+ @Override
+ public MetricsRecordBuilder add(AbstractMetric metric) {
+ add(metric.info(), metric.toString());
+ return this;
+ }
+
+ @Override
+ public MetricsRecordBuilder setContext(String value) {
+ return tuple("context", value);
+ }
+
+ @Override
+ public MetricsRecordBuilder addCounter(MetricsInfo info, int value) {
+ return add(info, value);
+ }
+
+ @Override
+ public MetricsRecordBuilder addCounter(MetricsInfo info, long value) {
+ return add(info, value);
+ }
+
+ @Override
+ public MetricsRecordBuilder addGauge(MetricsInfo info, int value) {
+ return add(info, value);
+ }
+
+ @Override
+ public MetricsRecordBuilder addGauge(MetricsInfo info, long value) {
+ return add(info, value);
+ }
+
+ @Override
+ public MetricsRecordBuilder addGauge(MetricsInfo info, float value) {
+ return add(info, value);
+ }
+
+ @Override
+ public MetricsRecordBuilder addGauge(MetricsInfo info, double value) {
+ return add(info, value);
+ }
+
+ @Override
+ public MetricsCollector parent() {
+ return parent;
+ }
+
+ @Override
+ public String toString() {
+ return builder.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableCounterLong.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableCounterLong.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableCounterLong.java
index 03a6043..d3dec2e 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableCounterLong.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableCounterLong.java
@@ -34,7 +34,7 @@ public class MutableCounterLong extends MutableCounter {
private AtomicLong value = new AtomicLong();
- MutableCounterLong(MetricsInfo info, long initValue) {
+ public MutableCounterLong(MetricsInfo info, long initValue) {
super(info);
this.value.set(initValue);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index bd061c9..2b13133 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -938,7 +938,15 @@
uploading (fs.s3a.threads.max) or queueing (fs.s3a.max.total.tasks)</description>
</property>
- <property>
+<property>
+ <name>fs.s3a.readahead.range</name>
+ <value>65536</value>
+ <description>Bytes to read ahead during a seek() before closing and
+ re-opening the S3 HTTP connection. This option will be overridden if
+ any call to setReadahead() is made to an open stream.</description>
+</property>
+
+<property>
<name>fs.s3a.fast.buffer.size</name>
<value>1048576</value>
<description>Size of initial memory buffer in bytes allocated for an
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml b/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml
index 204e6ab..2b4160a 100644
--- a/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml
+++ b/hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml
@@ -15,361 +15,8 @@
limitations under the License.
-->
<FindBugsFilter>
- <Match>
- <Package name="org.apache.hadoop.security.proto" />
- </Match>
- <Match>
- <Package name="org.apache.hadoop.tools.proto" />
- </Match>
- <Match>
- <Bug pattern="EI_EXPOSE_REP" />
- </Match>
- <Match>
- <Bug pattern="EI_EXPOSE_REP2" />
- </Match>
- <Match>
- <Bug pattern="SE_COMPARATOR_SHOULD_BE_SERIALIZABLE" />
- </Match>
- <Match>
- <Class name="~.*_jsp" />
- <Bug pattern="DLS_DEAD_LOCAL_STORE" />
- </Match>
- <Match>
- <Field name="_jspx_dependants" />
- <Bug pattern="UWF_UNWRITTEN_FIELD" />
- </Match>
- <!--
- Inconsistent synchronization for Client.Connection.out is
- is intentional to make a connection to be closed instantly.
- -->
- <Match>
- <Class name="org.apache.hadoop.ipc.Client$Connection" />
- <Field name="out" />
- <Bug pattern="IS2_INCONSISTENT_SYNC" />
- </Match>
- <!--
- Further SaslException should be ignored during cleanup and
- original exception should be re-thrown.
- -->
- <Match>
- <Class name="org.apache.hadoop.security.SaslRpcClient" />
- <Bug pattern="DE_MIGHT_IGNORE" />
- </Match>
- <!--
- Ignore Cross Scripting Vulnerabilities
- -->
- <Match>
- <Package name="~org.apache.hadoop.mapred.*" />
- <Bug code="XSS" />
- </Match>
- <Match>
- <Class name="org.apache.hadoop.mapred.taskdetails_jsp" />
- <Bug code="HRS" />
- </Match>
- <!--
- Ignore warnings where child class has the same name as
- super class. Classes based on Old API shadow names from
- new API. Should go off after HADOOP-1.0
- -->
- <Match>
- <Class name="~org.apache.hadoop.mapred.*" />
- <Bug pattern="NM_SAME_SIMPLE_NAME_AS_SUPERCLASS" />
- </Match>
- <!--
- Ignore warnings for usage of System.exit. This is
- required and have been well thought out
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.Child$2" />
- <Method name="run" />
- <Bug pattern="DM_EXIT" />
- </Match>
- <Match>
- <Class name="org.apache.hadoop.mapred.JobTracker" />
- <Method name="addHostToNodeMapping" />
- <Bug pattern="DM_EXIT" />
- </Match>
- <Match>
- <Class name="org.apache.hadoop.mapred.Task" />
- <Or>
- <Method name="done" />
- <Method name="commit" />
- <Method name="statusUpdate" />
- </Or>
- <Bug pattern="DM_EXIT" />
- </Match>
- <Match>
- <Class name="org.apache.hadoop.mapred.Task$TaskReporter" />
- <Method name="run" />
- <Bug pattern="DM_EXIT" />
- </Match>
- <Match>
- <Class name="org.apache.hadoop.util.ProgramDriver" />
- <Method name="driver" />
- <Bug pattern="DM_EXIT" />
- </Match>
- <Match>
- <Class name="org.apache.hadoop.util.RunJar" />
- <Method name="run" />
- <Bug pattern="DM_EXIT" />
- </Match>
- <!--
- We need to cast objects between old and new api objects
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.OutputCommitter" />
- <Bug pattern="BC_UNCONFIRMED_CAST" />
- </Match>
- <!--
- We intentionally do the get name from the inner class
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.TaskTracker$MapEventsFetcherThread" />
- <Method name="run" />
- <Bug pattern="IA_AMBIGUOUS_INVOCATION_OF_INHERITED_OR_OUTER_METHOD" />
- </Match>
- <Match>
- <Class name="org.apache.hadoop.mapred.FileOutputCommitter" />
- <Bug pattern="NM_WRONG_PACKAGE_INTENTIONAL" />
- </Match>
- <!--
- Ignoring this warning as resolving this would need a non-trivial change in code
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.lib.aggregate.ValueAggregatorBaseDescriptor" />
- <Method name="configure" />
- <Field name="maxNumItems" />
- <Bug pattern="ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD" />
- </Match>
- <!--
- Comes from org.apache.jasper.runtime.ResourceInjector. Cannot do much.
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.jobqueue_005fdetails_jsp" />
- <Field name="_jspx_resourceInjector" />
- <Bug pattern="SE_BAD_FIELD" />
- </Match>
- <!--
- Storing textInputFormat and then passing it as a parameter. Safe to ignore.
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.lib.aggregate.ValueAggregatorJob" />
- <Method name="createValueAggregatorJob" />
- <Bug pattern="DLS_DEAD_STORE_OF_CLASS_LITERAL" />
- </Match>
- <!--
- Can remove this after the upgrade to findbugs1.3.8
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.lib.db.DBInputFormat" />
- <Method name="getSplits" />
- <Bug pattern="DLS_DEAD_LOCAL_STORE" />
- </Match>
- <!--
- This is a spurious warning. Just ignore
- -->
- <Match>
- <Class name="org.apache.hadoop.mapred.MapTask$MapOutputBuffer" />
- <Field name="kvindex" />
- <Bug pattern="IS2_INCONSISTENT_SYNC" />
- </Match>
-
- <!--
- core changes
- -->
- <Match>
- <Class name="~org.apache.hadoop.*" />
- <Bug code="MS" />
- </Match>
-
- <Match>
- <Class name="org.apache.hadoop.fs.FileSystem" />
- <Method name="checkPath" />
- <Bug pattern="ES_COMPARING_STRINGS_WITH_EQ" />
- </Match>
-
- <Match>
- <Class name="org.apache.hadoop.io.Closeable" />
- <Bug pattern="NM_SAME_SIMPLE_NAME_AS_INTERFACE" />
- </Match>
-
- <Match>
- <Class name="org.apache.hadoop.security.AccessControlException" />
- <Bug pattern="NM_SAME_SIMPLE_NAME_AS_SUPERCLASS" />
- </Match>
-
- <Match>
- <Class name="org.apache.hadoop.util.ProcfsBasedProcessTree" />
- <Bug pattern="DMI_HARDCODED_ABSOLUTE_FILENAME" />
- </Match>
-
- <!--
- Streaming, Examples
- -->
- <Match>
- <Class name="org.apache.hadoop.streaming.StreamUtil$TaskId" />
- <Bug pattern="URF_UNREAD_FIELD" />
- </Match>
-
- <Match>
- <Class name="org.apache.hadoop.examples.DBCountPageView" />
- <Method name="verify" />
- <Bug pattern="OBL_UNSATISFIED_OBLIGATION" />
- </Match>
-
- <Match>
- <Class name="org.apache.hadoop.examples.ContextFactory" />
- <Method name="setAttributes" />
- <Bug pattern="OBL_UNSATISFIED_OBLIGATION" />
- </Match>
-
- <!--
- TFile
- -->
- <Match>
- <Class name="org.apache.hadoop.io.file.tfile.Chunk$ChunkDecoder" />
- <Method name="close" />
- <Bug pattern="SR_NOT_CHECKED" />
- </Match>
- <!--
- The purpose of skip() is to drain remaining bytes of the chunk-encoded
- stream (one chunk at a time). The termination condition is checked by
- checkEOF().
- -->
- <Match>
- <Class name="org.apache.hadoop.io.file.tfile.Utils" />
- <Method name="writeVLong" />
- <Bug pattern="SF_SWITCH_FALLTHROUGH" />
- </Match>
- <!--
- The switch condition fall through is intentional and for performance
- purposes.
- -->
-
- <Match>
- <Class name="org.apache.hadoop.log.EventCounter"/>
- <!-- backward compatibility -->
- <Bug pattern="NM_SAME_SIMPLE_NAME_AS_SUPERCLASS"/>
- </Match>
- <Match>
- <Class name="org.apache.hadoop.metrics.jvm.EventCounter"/>
- <!-- backward compatibility -->
- <Bug pattern="NM_SAME_SIMPLE_NAME_AS_SUPERCLASS"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ipc\.protobuf\.ProtobufRpcEngineProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ipc\.protobuf\.ProtocolInfoProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ipc\.protobuf\.IpcConnectionContextProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ipc\.protobuf\.RpcHeaderProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ha\.proto\.HAServiceProtocolProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ha\.proto\.ZKFCProtocolProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.security\.proto\.SecurityProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ipc\.protobuf\.TestProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ipc\.proto\.RefreshCallQueueProtocolProtos.*"/>
- </Match>
- <Match>
- <!-- protobuf generated code -->
- <Class name="~org\.apache\.hadoop\.ipc\.proto\.GenericRefreshProtocolProtos.*"/>
- </Match>
-
- <!--
- Manually checked, misses child thread manually syncing on parent's intrinsic lock.
- -->
- <Match>
- <Class name="org.apache.hadoop.metrics2.lib.MutableQuantiles" />
- <Field name="previousSnapshot" />
- <Bug pattern="IS2_INCONSISTENT_SYNC" />
- </Match>
- <!--
- The method uses a generic type T that extends two other types
- T1 and T2. Findbugs complains of a cast from T1 to T2.
- -->
- <Match>
- <Class name="org.apache.hadoop.fs.DelegationTokenRenewer" />
- <Method name="removeRenewAction" />
- <Bug pattern="BC_UNCONFIRMED_CAST" />
- </Match>
-
- <!-- Inconsistent synchronization flagged by findbugs is not valid. -->
- <Match>
- <Class name="org.apache.hadoop.ipc.Client$Connection" />
- <Field name="in" />
- <Bug pattern="IS2_INCONSISTENT_SYNC" />
- </Match>
- <!--
- The switch condition for INITIATE is expected to fallthru to RESPONSE
- to process initial sasl response token included in the INITIATE
- -->
- <Match>
- <Class name="org.apache.hadoop.ipc.Server$Connection" />
- <Method name="processSaslMessage" />
- <Bug pattern="SF_SWITCH_FALLTHROUGH" />
- </Match>
-
- <!-- Synchronization performed on util.concurrent instance. -->
- <Match>
- <Class name="org.apache.hadoop.service.AbstractService" />
- <Method name="stop" />
- <Bug code="JLM" />
- </Match>
-
- <Match>
- <Class name="org.apache.hadoop.service.AbstractService" />
- <Method name="waitForServiceToStop" />
- <Bug code="JLM" />
- </Match>
-
- <!--
- OpenStack Swift FS module -closes streams in a different method
- from where they are opened.
- -->
- <Match>
- <Class name="org.apache.hadoop.fs.swift.snative.SwiftNativeOutputStream"/>
- <Method name="uploadFileAttempt"/>
- <Bug pattern="OBL_UNSATISFIED_OBLIGATION"/>
- </Match>
- <Match>
- <Class name="org.apache.hadoop.fs.swift.snative.SwiftNativeOutputStream"/>
- <Method name="uploadFilePartAttempt"/>
- <Bug pattern="OBL_UNSATISFIED_OBLIGATION"/>
- </Match>
-
- <!-- code from maven source, null value is checked at callee side. -->
- <Match>
- <Class name="org.apache.hadoop.util.ComparableVersion$ListItem" />
- <Method name="compareTo" />
- <Bug code="NP" />
- </Match>
-
+ <!-- S3n warnings about malicious code aren't that relevant given its limited future. -->
<Match>
- <Class name="org.apache.hadoop.util.HttpExceptionUtils"/>
- <Method name="validateResponse"/>
- <Bug pattern="REC_CATCH_EXCEPTION"/>
+ <Class name="org.apache.hadoop.fs.s3.INode" />
</Match>
-
</FindBugsFilter>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/FileSystemStore.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/FileSystemStore.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/FileSystemStore.java
index 07e456b..3c7ed60 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/FileSystemStore.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/FileSystemStore.java
@@ -55,13 +55,13 @@ public interface FileSystemStore {
/**
* Delete everything. Used for testing.
- * @throws IOException
+ * @throws IOException on any problem
*/
void purge() throws IOException;
/**
* Diagnostic method to dump all INodes to the console.
- * @throws IOException
+ * @throws IOException on any problem
*/
void dump() throws IOException;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/S3Credentials.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/S3Credentials.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/S3Credentials.java
index fdacc3f..5ab352a 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/S3Credentials.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3/S3Credentials.java
@@ -38,6 +38,8 @@ public class S3Credentials {
private String secretAccessKey;
/**
+ * @param uri bucket URI optionally containing username and password.
+ * @param conf configuration
* @throws IllegalArgumentException if credentials for S3 cannot be
* determined.
* @throws IOException if credential providers are misconfigured and we have
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AnonymousAWSCredentialsProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AnonymousAWSCredentialsProvider.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AnonymousAWSCredentialsProvider.java
index 2a24273..e62ec77 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AnonymousAWSCredentialsProvider.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AnonymousAWSCredentialsProvider.java
@@ -21,7 +21,11 @@ package org.apache.hadoop.fs.s3a;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.AnonymousAWSCredentials;
import com.amazonaws.auth.AWSCredentials;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+@InterfaceAudience.Private
+@InterfaceStability.Stable
public class AnonymousAWSCredentialsProvider implements AWSCredentialsProvider {
public AWSCredentials getCredentials() {
return new AnonymousAWSCredentials();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/BasicAWSCredentialsProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/BasicAWSCredentialsProvider.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/BasicAWSCredentialsProvider.java
index 9a0adda..2f721e4 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/BasicAWSCredentialsProvider.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/BasicAWSCredentialsProvider.java
@@ -23,7 +23,11 @@ import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.auth.AWSCredentials;
import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+@InterfaceAudience.Private
+@InterfaceStability.Stable
public class BasicAWSCredentialsProvider implements AWSCredentialsProvider {
private final String accessKey;
private final String secretKey;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index bf3f85f..a800082 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -18,7 +18,19 @@
package org.apache.hadoop.fs.s3a;
-public class Constants {
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * All the constants used with the {@link S3AFileSystem}.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public final class Constants {
+
+ private Constants() {
+ }
+
// s3 access key
public static final String ACCESS_KEY = "fs.s3a.access.key";
@@ -124,4 +136,8 @@ public class Constants {
public static final int S3A_DEFAULT_PORT = -1;
public static final String USER_AGENT_PREFIX = "fs.s3a.user.agent.prefix";
+
+ /** read ahead buffer size to prevent connection re-establishments. */
+ public static final String READAHEAD_RANGE = "fs.s3a.readahead.range";
+ public static final long DEFAULT_READAHEAD_RANGE = 64 * 1024;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFastOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFastOutputStream.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFastOutputStream.java
index 5558693..bf0c7b9 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFastOutputStream.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFastOutputStream.java
@@ -37,6 +37,7 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.util.Progressable;
@@ -64,6 +65,7 @@ import java.util.concurrent.ExecutorService;
* <p>
* Unstable: statistics and error handling might evolve
*/
+@InterfaceAudience.Private
@InterfaceStability.Unstable
public class S3AFastOutputStream extends OutputStream {
@@ -102,7 +104,8 @@ public class S3AFastOutputStream extends OutputStream {
* @param partSize size of a single part in a multi-part upload (except
* last part)
* @param multiPartThreshold files at least this size use multi-part upload
- * @throws IOException
+ * @param threadPoolExecutor thread factory
+ * @throws IOException on any problem
*/
public S3AFastOutputStream(AmazonS3Client client, S3AFileSystem fs,
String bucket, String key, Progressable progress,
@@ -159,7 +162,7 @@ public class S3AFastOutputStream extends OutputStream {
* Writes a byte to the memory buffer. If this causes the buffer to reach
* its limit, the actual upload is submitted to the threadpool.
* @param b the int of which the lowest byte is written
- * @throws IOException
+ * @throws IOException on any problem
*/
@Override
public synchronized void write(int b) throws IOException {
@@ -177,10 +180,10 @@ public class S3AFastOutputStream extends OutputStream {
* @param b byte array containing
* @param off offset in array where to start
* @param len number of bytes to be written
- * @throws IOException
+ * @throws IOException on any problem
*/
@Override
- public synchronized void write(byte b[], int off, int len)
+ public synchronized void write(byte[] b, int off, int len)
throws IOException {
if (b == null) {
throw new NullPointerException();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
index 47caea8..9ecca33 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileStatus.java
@@ -17,9 +17,19 @@
*/
package org.apache.hadoop.fs.s3a;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
+/**
+ * File status for an S3A "file".
+ * Modification time is trouble, see {@link #getModificationTime()}.
+ *
+ * The subclass is private as it should not be created directly.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
public class S3AFileStatus extends FileStatus {
private boolean isEmptyDirectory;
@@ -45,7 +55,7 @@ public class S3AFileStatus extends FileStatus {
return System.getProperty("user.name");
}
- /** Compare if this object is equal to another object
+ /** Compare if this object is equal to another object.
* @param o the object to be compared.
* @return true if two file status has the same path name; false if not.
*/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/27c4e90e/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index 287474c..3f9723d 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -22,6 +22,7 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
+import java.io.InterruptedIOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Date;
@@ -56,8 +57,11 @@ import com.amazonaws.event.ProgressListener;
import com.amazonaws.event.ProgressEvent;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -76,9 +80,24 @@ import static org.apache.hadoop.fs.s3a.Constants.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * The core S3A Filesystem implementation.
+ *
+ * This subclass is marked as private as code should not be creating it
+ * directly; use {@link FileSystem#get(Configuration)} and variants to
+ * create one.
+ *
+ * If cast to {@code S3AFileSystem}, extra methods and features may be accessed.
+ * Consider those private and unstable.
+ *
+ * Because it prints some of the state of the instrumentation,
+ * the output of {@link #toString()} must also be considered unstable.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
public class S3AFileSystem extends FileSystem {
/**
- * Default blocksize as used in blocksize and FS status queries
+ * Default blocksize as used in blocksize and FS status queries.
*/
public static final int DEFAULT_BLOCKSIZE = 32 * 1024 * 1024;
private URI uri;
@@ -94,6 +113,8 @@ public class S3AFileSystem extends FileSystem {
public static final Logger LOG = LoggerFactory.getLogger(S3AFileSystem.class);
private CannedAccessControlList cannedACL;
private String serverSideEncryptionAlgorithm;
+ private S3AInstrumentation instrumentation;
+ private long readAhead;
// The maximum number of entries that can be deleted in any call to s3
private static final int MAX_ENTRIES_TO_DELETE = 1000;
@@ -105,10 +126,12 @@ public class S3AFileSystem extends FileSystem {
*/
public void initialize(URI name, Configuration conf) throws IOException {
super.initialize(name, conf);
+ setConf(conf);
+ instrumentation = new S3AInstrumentation(name);
uri = URI.create(name.getScheme() + "://" + name.getAuthority());
- workingDir = new Path("/user", System.getProperty("user.name")).makeQualified(this.uri,
- this.getWorkingDirectory());
+ workingDir = new Path("/user", System.getProperty("user.name"))
+ .makeQualified(this.uri, this.getWorkingDirectory());
AWSAccessKeys creds = getAWSAccessKeys(name, conf);
@@ -122,19 +145,20 @@ public class S3AFileSystem extends FileSystem {
bucket = name.getHost();
ClientConfiguration awsConf = new ClientConfiguration();
- awsConf.setMaxConnections(conf.getInt(MAXIMUM_CONNECTIONS,
- DEFAULT_MAXIMUM_CONNECTIONS));
+ awsConf.setMaxConnections(intOption(conf, MAXIMUM_CONNECTIONS,
+ DEFAULT_MAXIMUM_CONNECTIONS, 1));
boolean secureConnections = conf.getBoolean(SECURE_CONNECTIONS,
DEFAULT_SECURE_CONNECTIONS);
awsConf.setProtocol(secureConnections ? Protocol.HTTPS : Protocol.HTTP);
- awsConf.setMaxErrorRetry(conf.getInt(MAX_ERROR_RETRIES,
- DEFAULT_MAX_ERROR_RETRIES));
- awsConf.setConnectionTimeout(conf.getInt(ESTABLISH_TIMEOUT,
- DEFAULT_ESTABLISH_TIMEOUT));
- awsConf.setSocketTimeout(conf.getInt(SOCKET_TIMEOUT,
- DEFAULT_SOCKET_TIMEOUT));
+ awsConf.setMaxErrorRetry(intOption(conf, MAX_ERROR_RETRIES,
+ DEFAULT_MAX_ERROR_RETRIES, 0));
+ awsConf.setConnectionTimeout(intOption(conf, ESTABLISH_TIMEOUT,
+ DEFAULT_ESTABLISH_TIMEOUT, 0));
+ awsConf.setSocketTimeout(intOption(conf, SOCKET_TIMEOUT,
+ DEFAULT_SOCKET_TIMEOUT, 0));
String signerOverride = conf.getTrimmed(SIGNING_ALGORITHM, "");
- if(!signerOverride.isEmpty()) {
+ if (!signerOverride.isEmpty()) {
+ LOG.debug("Signer override = {}", signerOverride);
awsConf.setSignerOverride(signerOverride);
}
@@ -144,21 +168,23 @@ public class S3AFileSystem extends FileSystem {
initAmazonS3Client(conf, credentials, awsConf);
- maxKeys = conf.getInt(MAX_PAGING_KEYS, DEFAULT_MAX_PAGING_KEYS);
+ maxKeys = intOption(conf, MAX_PAGING_KEYS, DEFAULT_MAX_PAGING_KEYS, 1);
partSize = conf.getLong(MULTIPART_SIZE, DEFAULT_MULTIPART_SIZE);
- multiPartThreshold = conf.getLong(MIN_MULTIPART_THRESHOLD,
- DEFAULT_MIN_MULTIPART_THRESHOLD);
- enableMultiObjectsDelete = conf.getBoolean(ENABLE_MULTI_DELETE, true);
-
if (partSize < 5 * 1024 * 1024) {
LOG.error(MULTIPART_SIZE + " must be at least 5 MB");
partSize = 5 * 1024 * 1024;
}
+ multiPartThreshold = conf.getLong(MIN_MULTIPART_THRESHOLD,
+ DEFAULT_MIN_MULTIPART_THRESHOLD);
if (multiPartThreshold < 5 * 1024 * 1024) {
LOG.error(MIN_MULTIPART_THRESHOLD + " must be at least 5 MB");
multiPartThreshold = 5 * 1024 * 1024;
}
+ //check but do not store the block size
+ longOption(conf, FS_S3A_BLOCK_SIZE, DEFAULT_BLOCKSIZE, 1);
+ enableMultiObjectsDelete = conf.getBoolean(ENABLE_MULTI_DELETE, true);
+ readAhead = longOption(conf, READAHEAD_RANGE, DEFAULT_READAHEAD_RANGE, 0);
int maxThreads = conf.getInt(MAX_THREADS, DEFAULT_MAX_THREADS);
if (maxThreads < 2) {
@@ -180,19 +206,17 @@ public class S3AFileSystem extends FileSystem {
initCannedAcls(conf);
if (!s3.doesBucketExist(bucket)) {
- throw new IOException("Bucket " + bucket + " does not exist");
+ throw new FileNotFoundException("Bucket " + bucket + " does not exist");
}
initMultipartUploads(conf);
serverSideEncryptionAlgorithm = conf.get(SERVER_SIDE_ENCRYPTION_ALGORITHM);
- setConf(conf);
}
void initProxySupport(Configuration conf, ClientConfiguration awsConf,
- boolean secureConnections) throws IllegalArgumentException,
- IllegalArgumentException {
+ boolean secureConnections) throws IllegalArgumentException {
String proxyHost = conf.getTrimmed(PROXY_HOST, "");
int proxyPort = conf.getInt(PROXY_PORT, -1);
if (!proxyHost.isEmpty()) {
@@ -223,7 +247,8 @@ public class S3AFileSystem extends FileSystem {
if (LOG.isDebugEnabled()) {
LOG.debug("Using proxy server {}:{} as user {} with password {} on " +
"domain {} as workstation {}", awsConf.getProxyHost(),
- awsConf.getProxyPort(), String.valueOf(awsConf.getProxyUsername()),
+ awsConf.getProxyPort(),
+ String.valueOf(awsConf.getProxyUsername()),
awsConf.getProxyPassword(), awsConf.getProxyDomain(),
awsConf.getProxyWorkstation());
}
@@ -258,7 +283,7 @@ public class S3AFileSystem extends FileSystem {
AWSCredentialsProviderChain credentials, ClientConfiguration awsConf)
throws IllegalArgumentException {
s3 = new AmazonS3Client(credentials, awsConf);
- String endPoint = conf.getTrimmed(ENDPOINT,"");
+ String endPoint = conf.getTrimmed(ENDPOINT, "");
if (!endPoint.isEmpty()) {
try {
s3.setEndpoint(endPoint);
@@ -301,14 +326,25 @@ public class S3AFileSystem extends FileSystem {
private void initMultipartUploads(Configuration conf) {
boolean purgeExistingMultipart = conf.getBoolean(PURGE_EXISTING_MULTIPART,
- DEFAULT_PURGE_EXISTING_MULTIPART);
- long purgeExistingMultipartAge = conf.getLong(PURGE_EXISTING_MULTIPART_AGE,
- DEFAULT_PURGE_EXISTING_MULTIPART_AGE);
+ DEFAULT_PURGE_EXISTING_MULTIPART);
+ long purgeExistingMultipartAge = longOption(conf,
+ PURGE_EXISTING_MULTIPART_AGE, DEFAULT_PURGE_EXISTING_MULTIPART_AGE, 0);
if (purgeExistingMultipart) {
- Date purgeBefore = new Date(new Date().getTime() - purgeExistingMultipartAge*1000);
+ Date purgeBefore =
+ new Date(new Date().getTime() - purgeExistingMultipartAge * 1000);
- transfers.abortMultipartUploads(bucket, purgeBefore);
+ try {
+ transfers.abortMultipartUploads(bucket, purgeBefore);
+ } catch (AmazonServiceException e) {
+ if (e.getStatusCode() == 403) {
+ instrumentation.errorIgnored();
+ LOG.debug("Failed to abort multipart uploads against {}," +
+ " FS may be read only", bucket, e);
+ } else {
+ throw e;
+ }
+ }
}
}
@@ -421,16 +457,15 @@ public class S3AFileSystem extends FileSystem {
public FSDataInputStream open(Path f, int bufferSize)
throws IOException {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Opening '{}' for reading.", f);
- }
+ LOG.debug("Opening '{}' for reading.", f);
final FileStatus fileStatus = getFileStatus(f);
if (fileStatus.isDirectory()) {
- throw new FileNotFoundException("Can't open " + f + " because it is a directory");
+ throw new FileNotFoundException("Can't open " + f
+ + " because it is a directory");
}
return new FSDataInputStream(new S3AInputStream(bucket, pathToKey(f),
- fileStatus.getLen(), s3, statistics));
+ fileStatus.getLen(), s3, statistics, instrumentation, readAhead));
}
/**
@@ -456,16 +491,26 @@ public class S3AFileSystem extends FileSystem {
if (!overwrite && exists(f)) {
throw new FileAlreadyExistsException(f + " already exists");
}
+ instrumentation.fileCreated();
if (getConf().getBoolean(FAST_UPLOAD, DEFAULT_FAST_UPLOAD)) {
return new FSDataOutputStream(new S3AFastOutputStream(s3, this, bucket,
key, progress, statistics, cannedACL,
serverSideEncryptionAlgorithm, partSize, multiPartThreshold,
threadPoolExecutor), statistics);
}
- // We pass null to FSDataOutputStream so it won't count writes that are being buffered to a file
- return new FSDataOutputStream(new S3AOutputStream(getConf(), transfers, this,
- bucket, key, progress, cannedACL, statistics,
- serverSideEncryptionAlgorithm), null);
+ // We pass null to FSDataOutputStream so it won't count writes that
+ // are being buffered to a file
+ return new FSDataOutputStream(
+ new S3AOutputStream(getConf(),
+ transfers,
+ this,
+ bucket,
+ key,
+ progress,
+ cannedACL,
+ statistics,
+ serverSideEncryptionAlgorithm),
+ null);
}
/**
@@ -476,7 +521,7 @@ public class S3AFileSystem extends FileSystem {
* @throws IOException indicating that append is not supported.
*/
public FSDataOutputStream append(Path f, int bufferSize,
- Progressable progress) throws IOException {
+ Progressable progress) throws IOException {
throw new IOException("Not supported");
}
@@ -501,17 +546,13 @@ public class S3AFileSystem extends FileSystem {
* @return true if rename is successful
*/
public boolean rename(Path src, Path dst) throws IOException {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Rename path {} to {}", src, dst);
- }
+ LOG.debug("Rename path {} to {}", src, dst);
String srcKey = pathToKey(src);
String dstKey = pathToKey(dst);
if (srcKey.isEmpty() || dstKey.isEmpty()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("rename: src or dst are empty");
- }
+ LOG.debug("rename: source {} or dest {}, is empty", srcKey, dstKey);
return false;
}
@@ -524,9 +565,8 @@ public class S3AFileSystem extends FileSystem {
}
if (srcKey.equals(dstKey)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("rename: src and dst refer to the same file or directory");
- }
+ LOG.debug("rename: src and dst refer to the same file or directory: {}",
+ dst);
return srcStatus.isFile();
}
@@ -535,9 +575,8 @@ public class S3AFileSystem extends FileSystem {
dstStatus = getFileStatus(dst);
if (srcStatus.isDirectory() && dstStatus.isFile()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("rename: src is a directory and dst is a file");
- }
+ LOG.debug("rename: src {} is a directory and dst {} is a file",
+ src, dst);
return false;
}
@@ -545,6 +584,7 @@ public class S3AFileSystem extends FileSystem {
return false;
}
} catch (FileNotFoundException e) {
+ LOG.debug("rename: destination path {} not found", dst);
// Parent must exist
Path parent = dst.getParent();
if (!pathToKey(parent).isEmpty()) {
@@ -554,6 +594,8 @@ public class S3AFileSystem extends FileSystem {
return false;
}
} catch (FileNotFoundException e2) {
+ LOG.debug("rename: destination path {} has no parent {}",
+ dst, parent);
return false;
}
}
@@ -561,9 +603,7 @@ public class S3AFileSystem extends FileSystem {
// Ok! Time to start
if (srcStatus.isFile()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("rename: renaming file " + src + " to " + dst);
- }
+ LOG.debug("rename: renaming file {} to {}", src, dst);
if (dstStatus != null && dstStatus.isDirectory()) {
String newDstKey = dstKey;
if (!newDstKey.endsWith("/")) {
@@ -572,15 +612,13 @@ public class S3AFileSystem extends FileSystem {
String filename =
srcKey.substring(pathToKey(src.getParent()).length()+1);
newDstKey = newDstKey + filename;
- copyFile(srcKey, newDstKey);
+ copyFile(srcKey, newDstKey, srcStatus.getLen());
} else {
- copyFile(srcKey, dstKey);
+ copyFile(srcKey, dstKey, srcStatus.getLen());
}
delete(src, false);
} else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("rename: renaming directory " + src + " to " + dst);
- }
+ LOG.debug("rename: renaming directory {} to {}", src, dst);
// This is a directory to directory copy
if (!dstKey.endsWith("/")) {
@@ -593,14 +631,12 @@ public class S3AFileSystem extends FileSystem {
//Verify dest is not a child of the source directory
if (dstKey.startsWith(srcKey)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("cannot rename a directory to a subdirectory of self");
- }
+ LOG.debug("cannot rename a directory {}" +
+ " to a subdirectory of self: {}", srcKey, dstKey);
return false;
}
- List<DeleteObjectsRequest.KeyVersion> keysToDelete =
- new ArrayList<>();
+ List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
if (dstStatus != null && dstStatus.isEmptyDirectory()) {
// delete unnecessary fake directory.
keysToDelete.add(new DeleteObjectsRequest.KeyVersion(dstKey));
@@ -618,7 +654,7 @@ public class S3AFileSystem extends FileSystem {
for (S3ObjectSummary summary : objects.getObjectSummaries()) {
keysToDelete.add(new DeleteObjectsRequest.KeyVersion(summary.getKey()));
String newDstKey = dstKey + summary.getKey().substring(srcKey.length());
- copyFile(summary.getKey(), newDstKey);
+ copyFile(summary.getKey(), newDstKey, summary.getSize());
if (keysToDelete.size() == MAX_ENTRIES_TO_DELETE) {
removeKeys(keysToDelete, true);
@@ -657,6 +693,7 @@ public class S3AFileSystem extends FileSystem {
DeleteObjectsRequest deleteRequest
= new DeleteObjectsRequest(bucket).withKeys(keysToDelete);
s3.deleteObjects(deleteRequest);
+ instrumentation.fileDeleted(keysToDelete.size());
statistics.incrementWriteOps(1);
} else {
int writeops = 0;
@@ -666,7 +703,7 @@ public class S3AFileSystem extends FileSystem {
new DeleteObjectRequest(bucket, keyVersion.getKey()));
writeops++;
}
-
+ instrumentation.fileDeleted(keysToDelete.size());
statistics.incrementWriteOps(writeops);
}
if (clearKeys) {
@@ -684,25 +721,20 @@ public class S3AFileSystem extends FileSystem {
* @throws IOException due to inability to delete a directory or file.
*/
public boolean delete(Path f, boolean recursive) throws IOException {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Delete path " + f + " - recursive " + recursive);
- }
+ LOG.debug("Delete path {} - recursive {}", f , recursive);
S3AFileStatus status;
try {
status = getFileStatus(f);
} catch (FileNotFoundException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Couldn't delete " + f + " - does not exist");
- }
+ LOG.debug("Couldn't delete {} - does not exist", f);
+ instrumentation.errorIgnored();
return false;
}
String key = pathToKey(f);
if (status.isDirectory()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("delete: Path is a directory");
- }
+ LOG.debug("delete: Path is a directory: {}", f);
if (!recursive && !status.isEmptyDirectory()) {
throw new IOException("Path is a folder: " + f +
@@ -719,15 +751,12 @@ public class S3AFileSystem extends FileSystem {
}
if (status.isEmptyDirectory()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Deleting fake empty directory");
- }
+ LOG.debug("Deleting fake empty directory {}", key);
s3.deleteObject(bucket, key);
+ instrumentation.directoryDeleted();
statistics.incrementWriteOps(1);
} else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Getting objects for directory prefix " + key + " to delete");
- }
+ LOG.debug("Getting objects for directory prefix {} to delete", key);
ListObjectsRequest request = new ListObjectsRequest();
request.setBucketName(bucket);
@@ -736,16 +765,13 @@ public class S3AFileSystem extends FileSystem {
//request.setDelimiter("/");
request.setMaxKeys(maxKeys);
- List<DeleteObjectsRequest.KeyVersion> keys =
- new ArrayList<>();
+ List<DeleteObjectsRequest.KeyVersion> keys = new ArrayList<>();
ObjectListing objects = s3.listObjects(request);
statistics.incrementReadOps(1);
while (true) {
for (S3ObjectSummary summary : objects.getObjectSummaries()) {
keys.add(new DeleteObjectsRequest.KeyVersion(summary.getKey()));
- if (LOG.isDebugEnabled()) {
- LOG.debug("Got object to delete " + summary.getKey());
- }
+ LOG.debug("Got object to delete {}", summary.getKey());
if (keys.size() == MAX_ENTRIES_TO_DELETE) {
removeKeys(keys, true);
@@ -764,10 +790,9 @@ public class S3AFileSystem extends FileSystem {
}
}
} else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("delete: Path is a file");
- }
+ LOG.debug("delete: Path is a file");
s3.deleteObject(bucket, key);
+ instrumentation.fileDeleted(1);
statistics.incrementWriteOps(1);
}
@@ -779,9 +804,7 @@ public class S3AFileSystem extends FileSystem {
private void createFakeDirectoryIfNecessary(Path f) throws IOException {
String key = pathToKey(f);
if (!key.isEmpty() && !exists(f)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Creating new fake directory at " + f);
- }
+ LOG.debug("Creating new fake directory at {}", f);
createFakeDirectory(bucket, key);
}
}
@@ -798,9 +821,7 @@ public class S3AFileSystem extends FileSystem {
public FileStatus[] listStatus(Path f) throws FileNotFoundException,
IOException {
String key = pathToKey(f);
- if (LOG.isDebugEnabled()) {
- LOG.debug("List status for path: " + f);
- }
+ LOG.debug("List status for path: {}", f);
final List<FileStatus> result = new ArrayList<FileStatus>();
final FileStatus fileStatus = getFileStatus(f);
@@ -816,9 +837,7 @@ public class S3AFileSystem extends FileSystem {
request.setDelimiter("/");
request.setMaxKeys(maxKeys);
- if (LOG.isDebugEnabled()) {
- LOG.debug("listStatus: doing listObjects for directory " + key);
- }
+ LOG.debug("listStatus: doing listObjects for directory {}", key);
ObjectListing objects = s3.listObjects(request);
statistics.incrementReadOps(1);
@@ -831,24 +850,18 @@ public class S3AFileSystem extends FileSystem {
// Skip over keys that are ourselves and old S3N _$folder$ files
if (keyPath.equals(fQualified) ||
summary.getKey().endsWith(S3N_FOLDER_SUFFIX)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Ignoring: " + keyPath);
- }
+ LOG.debug("Ignoring: {}", keyPath);
continue;
}
if (objectRepresentsDirectory(summary.getKey(), summary.getSize())) {
result.add(new S3AFileStatus(true, true, keyPath));
- if (LOG.isDebugEnabled()) {
- LOG.debug("Adding: fd: " + keyPath);
- }
+ LOG.debug("Adding: fd: {}", keyPath);
} else {
result.add(new S3AFileStatus(summary.getSize(),
dateToLong(summary.getLastModified()), keyPath,
getDefaultBlockSize(fQualified)));
- if (LOG.isDebugEnabled()) {
- LOG.debug("Adding: fi: " + keyPath);
- }
+ LOG.debug("Adding: fi: {}", keyPath);
}
}
@@ -858,16 +871,11 @@ public class S3AFileSystem extends FileSystem {
continue;
}
result.add(new S3AFileStatus(true, false, keyPath));
- if (LOG.isDebugEnabled()) {
- LOG.debug("Adding: rd: " + keyPath);
- }
+ LOG.debug("Adding: rd: {}", keyPath);
}
if (objects.isTruncated()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("listStatus: list truncated - getting next batch");
- }
-
+ LOG.debug("listStatus: list truncated - getting next batch");
objects = s3.listNextBatchOfObjects(objects);
statistics.incrementReadOps(1);
} else {
@@ -875,9 +883,7 @@ public class S3AFileSystem extends FileSystem {
}
}
} else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Adding: rd (not a dir): " + f);
- }
+ LOG.debug("Adding: rd (not a dir): {}", f);
result.add(fileStatus);
}
@@ -890,14 +896,14 @@ public class S3AFileSystem extends FileSystem {
* Set the current working directory for the given file system. All relative
* paths will be resolved relative to it.
*
- * @param new_dir the current working directory.
+ * @param newDir the current working directory.
*/
- public void setWorkingDirectory(Path new_dir) {
- workingDir = new_dir;
+ public void setWorkingDirectory(Path newDir) {
+ workingDir = newDir;
}
/**
- * Get the current working directory for the given file system
+ * Get the current working directory for the given file system.
* @return the directory pathname
*/
public Path getWorkingDirectory() {
@@ -914,10 +920,7 @@ public class S3AFileSystem extends FileSystem {
// TODO: If we have created an empty file at /foo/bar and we then call
// mkdirs for /foo/bar/baz/roo what happens to the empty file /foo/bar/?
public boolean mkdirs(Path f, FsPermission permission) throws IOException {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Making directory: " + f);
- }
-
+ LOG.debug("Making directory: {}", f);
try {
FileStatus fileStatus = getFileStatus(f);
@@ -938,6 +941,7 @@ public class S3AFileSystem extends FileSystem {
fPart));
}
} catch (FileNotFoundException fnfe) {
+ instrumentation.errorIgnored();
}
fPart = fPart.getParent();
} while (fPart != null);
@@ -957,10 +961,7 @@ public class S3AFileSystem extends FileSystem {
*/
public S3AFileStatus getFileStatus(Path f) throws IOException {
String key = pathToKey(f);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Getting path status for " + f + " (" + key + ")");
- }
-
+ LOG.debug("Getting path status for {} ({})", f , key);
if (!key.isEmpty()) {
try {
@@ -968,15 +969,11 @@ public class S3AFileSystem extends FileSystem {
statistics.incrementReadOps(1);
if (objectRepresentsDirectory(key, meta.getContentLength())) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Found exact file: fake directory");
- }
+ LOG.debug("Found exact file: fake directory");
return new S3AFileStatus(true, true,
f.makeQualified(uri, workingDir));
} else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Found exact file: normal file");
- }
+ LOG.debug("Found exact file: normal file");
return new S3AFileStatus(meta.getContentLength(),
dateToLong(meta.getLastModified()),
f.makeQualified(uri, workingDir),
@@ -984,25 +981,23 @@ public class S3AFileSystem extends FileSystem {
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
- printAmazonServiceException(e);
+ printAmazonServiceException(f.toString(), e);
throw e;
}
} catch (AmazonClientException e) {
- printAmazonClientException(e);
+ printAmazonClientException(f.toString(), e);
throw e;
}
// Necessary?
if (!key.endsWith("/")) {
+ String newKey = key + "/";
try {
- String newKey = key + "/";
ObjectMetadata meta = s3.getObjectMetadata(bucket, newKey);
statistics.incrementReadOps(1);
if (objectRepresentsDirectory(newKey, meta.getContentLength())) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Found file (with /): fake directory");
- }
+ LOG.debug("Found file (with /): fake directory");
return new S3AFileStatus(true, true, f.makeQualified(uri, workingDir));
} else {
LOG.warn("Found file (with /): real file? should not happen: {}", key);
@@ -1014,11 +1009,11 @@ public class S3AFileSystem extends FileSystem {
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
- printAmazonServiceException(e);
+ printAmazonServiceException(newKey, e);
throw e;
}
} catch (AmazonClientException e) {
- printAmazonClientException(e);
+ printAmazonClientException(newKey, e);
throw e;
}
}
@@ -1038,17 +1033,17 @@ public class S3AFileSystem extends FileSystem {
statistics.incrementReadOps(1);
if (!objects.getCommonPrefixes().isEmpty()
- || objects.getObjectSummaries().size() > 0) {
+ || !objects.getObjectSummaries().isEmpty()) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Found path as directory (with /): " +
- objects.getCommonPrefixes().size() + "/" +
+ LOG.debug("Found path as directory (with /): {}/{}",
+ objects.getCommonPrefixes().size() ,
objects.getObjectSummaries().size());
for (S3ObjectSummary summary : objects.getObjectSummaries()) {
- LOG.debug("Summary: " + summary.getKey() + " " + summary.getSize());
+ LOG.debug("Summary: {} {}", summary.getKey(), summary.getSize());
}
for (String prefix : objects.getCommonPrefixes()) {
- LOG.debug("Prefix: " + prefix);
+ LOG.debug("Prefix: {}", prefix);
}
}
@@ -1060,17 +1055,15 @@ public class S3AFileSystem extends FileSystem {
}
} catch (AmazonServiceException e) {
if (e.getStatusCode() != 404) {
- printAmazonServiceException(e);
+ printAmazonServiceException(key, e);
throw e;
}
} catch (AmazonClientException e) {
- printAmazonClientException(e);
+ printAmazonClientException(key, e);
throw e;
}
- if (LOG.isDebugEnabled()) {
- LOG.debug("Not Found: " + f);
- }
+ LOG.debug("Not Found: {}", f);
throw new FileNotFoundException("No such file or directory: " + f);
}
@@ -1089,15 +1082,13 @@ public class S3AFileSystem extends FileSystem {
*/
@Override
public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path src,
- Path dst) throws IOException {
+ Path dst) throws IOException {
String key = pathToKey(dst);
if (!overwrite && exists(dst)) {
- throw new IOException(dst + " already exists");
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("Copying local file from " + src + " to " + dst);
+ throw new FileAlreadyExistsException(dst + " already exists");
}
+ LOG.debug("Copying local file from {} to {}", src, dst);
// Since we have a local file, we don't need to stream into a temporary file
LocalFileSystem local = getLocal(getConf());
@@ -1123,13 +1114,14 @@ public class S3AFileSystem extends FileSystem {
}
};
+ statistics.incrementWriteOps(1);
Upload up = transfers.upload(putObjectRequest);
up.addProgressListener(progressListener);
try {
up.waitForUploadResult();
- statistics.incrementWriteOps(1);
} catch (InterruptedException e) {
- throw new IOException("Got interrupted, cancelling");
+ throw new InterruptedIOException("Interrupted copying " + src
+ + " to " + dst + ", cancelling");
}
// This will delete unnecessary fake parent directories
@@ -1153,7 +1145,7 @@ public class S3AFileSystem extends FileSystem {
}
/**
- * Override getCononicalServiceName because we don't support token in S3A
+ * Override getCanonicalServiceName because we don't support token in S3A.
*/
@Override
public String getCanonicalServiceName() {
@@ -1161,17 +1153,17 @@ public class S3AFileSystem extends FileSystem {
return null;
}
- private void copyFile(String srcKey, String dstKey) throws IOException {
- if (LOG.isDebugEnabled()) {
- LOG.debug("copyFile " + srcKey + " -> " + dstKey);
- }
+ private void copyFile(String srcKey, String dstKey, long size)
+ throws IOException {
+ LOG.debug("copyFile {} -> {} ", srcKey, dstKey);
ObjectMetadata srcom = s3.getObjectMetadata(bucket, srcKey);
ObjectMetadata dstom = cloneObjectMetadata(srcom);
if (StringUtils.isNotBlank(serverSideEncryptionAlgorithm)) {
dstom.setSSEAlgorithm(serverSideEncryptionAlgorithm);
}
- CopyObjectRequest copyObjectRequest = new CopyObjectRequest(bucket, srcKey, bucket, dstKey);
+ CopyObjectRequest copyObjectRequest =
+ new CopyObjectRequest(bucket, srcKey, bucket, dstKey);
copyObjectRequest.setCannedAccessControlList(cannedACL);
copyObjectRequest.setNewObjectMetadata(dstom);
@@ -1192,13 +1184,17 @@ public class S3AFileSystem extends FileSystem {
try {
copy.waitForCopyResult();
statistics.incrementWriteOps(1);
+ instrumentation.filesCopied(1, size);
} catch (InterruptedException e) {
- throw new IOException("Got interrupted, cancelling");
+ throw new InterruptedIOException("Interrupted copying " + srcKey
+ + " to " + dstKey + ", cancelling");
}
}
private boolean objectRepresentsDirectory(final String name, final long size) {
- return !name.isEmpty() && name.charAt(name.length() - 1) == '/' && size == 0L;
+ return !name.isEmpty()
+ && name.charAt(name.length() - 1) == '/'
+ && size == 0L;
}
// Handles null Dates that can be returned by AWS
@@ -1216,8 +1212,9 @@ public class S3AFileSystem extends FileSystem {
private void deleteUnnecessaryFakeDirectories(Path f) throws IOException {
while (true) {
+ String key = "";
try {
- String key = pathToKey(f);
+ key = pathToKey(f);
if (key.isEmpty()) {
break;
}
@@ -1225,13 +1222,13 @@ public class S3AFileSystem extends FileSystem {
S3AFileStatus status = getFileStatus(f);
if (status.isDirectory() && status.isEmptyDirectory()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Deleting fake directory " + key + "/");
- }
+ LOG.debug("Deleting fake directory {}/", key);
s3.deleteObject(bucket, key + "/");
statistics.incrementWriteOps(1);
}
} catch (FileNotFoundException | AmazonServiceException e) {
+ LOG.debug("While deleting key {} ", key, e);
+ instrumentation.errorIgnored();
}
if (f.isRoot()) {
@@ -1267,10 +1264,12 @@ public class S3AFileSystem extends FileSystem {
if (StringUtils.isNotBlank(serverSideEncryptionAlgorithm)) {
om.setSSEAlgorithm(serverSideEncryptionAlgorithm);
}
- PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, objectName, im, om);
+ PutObjectRequest putObjectRequest =
+ new PutObjectRequest(bucketName, objectName, im, om);
putObjectRequest.setCannedAcl(cannedACL);
s3.putObject(putObjectRequest);
statistics.incrementWriteOps(1);
+ instrumentation.directoryCreated();
}
/**
@@ -1342,31 +1341,115 @@ public class S3AFileSystem extends FileSystem {
/**
* Return the number of bytes that large input files should be optimally
- * be split into to minimize i/o time.
+ * be split into to minimize I/O time.
* @deprecated use {@link #getDefaultBlockSize(Path)} instead
*/
@Deprecated
public long getDefaultBlockSize() {
- // default to 32MB: large enough to minimize the impact of seeks
return getConf().getLong(FS_S3A_BLOCK_SIZE, DEFAULT_BLOCKSIZE);
}
- private void printAmazonServiceException(AmazonServiceException ase) {
- LOG.info("Caught an AmazonServiceException, which means your request made it " +
- "to Amazon S3, but was rejected with an error response for some reason.");
- LOG.info("Error Message: " + ase.getMessage());
- LOG.info("HTTP Status Code: " + ase.getStatusCode());
- LOG.info("AWS Error Code: " + ase.getErrorCode());
- LOG.info("Error Type: " + ase.getErrorType());
- LOG.info("Request ID: " + ase.getRequestId());
- LOG.info("Class Name: " + ase.getClass().getName());
+ private void printAmazonServiceException(String target,
+ AmazonServiceException ase) {
+ LOG.info("{}: caught an AmazonServiceException {}", target, ase);
+ LOG.info("This means your request made it to Amazon S3," +
+ " but was rejected with an error response for some reason.");
+ LOG.info("Error Message: {}", ase.getMessage());
+ LOG.info("HTTP Status Code: {}", ase.getStatusCode());
+ LOG.info("AWS Error Code: {}", ase.getErrorCode());
+ LOG.info("Error Type: {}", ase.getErrorType());
+ LOG.info("Request ID: {}", ase.getRequestId());
+ LOG.info("Class Name: {}", ase.getClass().getName());
+ LOG.info("Exception", ase);
+ }
+
+ private void printAmazonClientException(String target,
+ AmazonClientException ace) {
+ LOG.info("{}: caught an AmazonClientException {}", target, ace);
+ LOG.info("This means the client encountered " +
+ "a problem while trying to communicate with S3, " +
+ "such as not being able to access the network.", ace);
}
- private void printAmazonClientException(AmazonClientException ace) {
- LOG.info("Caught an AmazonClientException, which means the client encountered " +
- "a serious internal problem while trying to communicate with S3, " +
- "such as not being able to access the network.");
- LOG.info("Error Message: {}" + ace, ace);
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder(
+ "S3AFileSystem{");
+ sb.append("uri=").append(uri);
+ sb.append(", workingDir=").append(workingDir);
+ sb.append(", partSize=").append(partSize);
+ sb.append(", enableMultiObjectsDelete=").append(enableMultiObjectsDelete);
+ sb.append(", maxKeys=").append(maxKeys);
+ sb.append(", cannedACL=").append(cannedACL.toString());
+ sb.append(", readAhead=").append(readAhead);
+ sb.append(", blockSize=").append(getDefaultBlockSize());
+ sb.append(", multiPartThreshold=").append(multiPartThreshold);
+ if (serverSideEncryptionAlgorithm != null) {
+ sb.append(", serverSideEncryptionAlgorithm='")
+ .append(serverSideEncryptionAlgorithm)
+ .append('\'');
+ }
+ sb.append(", statistics {")
+ .append(statistics.toString())
+ .append("}");
+ sb.append(", metrics {")
+ .append(instrumentation.dump("{", "=", "} ", true))
+ .append("}");
+ sb.append('}');
+ return sb.toString();
+ }
+
+ /**
+ * Get the partition size for multipart operations.
+ * @return the value as set during initialization
+ */
+ public long getPartitionSize() {
+ return partSize;
+ }
+
+ /**
+ * Get the threshold for multipart files
+ * @return the value as set during initialization
+ */
+ public long getMultiPartThreshold() {
+ return multiPartThreshold;
+ }
+
+ /**
+ * Get a integer option >= the minimum allowed value.
+ * @param conf configuration
+ * @param key key to look up
+ * @param defVal default value
+ * @param min minimum value
+ * @return the value
+ * @throws IllegalArgumentException if the value is below the minimum
+ */
+ static int intOption(Configuration conf, String key, int defVal, int min) {
+ int v = conf.getInt(key, defVal);
+ Preconditions.checkArgument(v >= min,
+ String.format("Value of %s: %d is below the minimum value %d",
+ key, v, min));
+ return v;
+ }
+
+ /**
+ * Get a long option >= the minimum allowed value.
+ * @param conf configuration
+ * @param key key to look up
+ * @param defVal default value
+ * @param min minimum value
+ * @return the value
+ * @throws IllegalArgumentException if the value is below the minimum
+ */
+ static long longOption(Configuration conf,
+ String key,
+ long defVal,
+ long min) {
+ long v = conf.getLong(key, defVal);
+ Preconditions.checkArgument(v >= min,
+ String.format("Value of %s: %d is below the minimum value %d",
+ key, v, min));
+ return v;
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org