[
https://issues.apache.org/jira/browse/HADOOP-13786?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16208734#comment-16208734
]
Aaron Fabbri commented on HADOOP-13786:
---------------------------------------
Going to start posting some comments on v39 megapatch. I still have to go back
over some of the more involved parts of the code.
{noformat}
+++ b/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
...
+ <!-- findbugs is mistaken -->
+ <Match>
+ <Class name="org.apache.hadoop.util.JsonSerializationy"/>
{noformat}
Elaboration would be nice. Also I think you've misspelled the class name.
{noformat}
+ * This method was copied from
+ * {@code org.apache.hadoop.registry.client.binding.JsonSerDeser}.
+ * @param <T> Type to marshal.
{noformat}
Can you explain why copy/paste versus import?
{noformat}
+
+<property>
+ <name>fs.s3a.retry.throttle.interval</name>
+ <value>1000ms</value>
{noformat}
Interesting default. Any reasoning? Feels high, subjectively, but that depends
on how throttling responses arrive at nodes in a cluster. If there is a large
batch of them spread across nodes I'd think smaller interval + randomization
would behave better (versus thundering herd after full second of silence).
{noformat}
+ public static PathOutputCommitterFactory getCommitterFactory(
...
+ if (StringUtils.isNotEmpty(conf.getTrimmed(schemeKey))) {
...
+ key = schemeKey;
+ }
+ } else {
+ // Either there's an explicit committer factory declared, or
+ // there's no output path, so trying to do schema-specific lookup
+ // cannot work.
+ }
{noformat}
Empty else block. Findbugs doesn't complain? Anyways you could just start
your comment with "Else: Either..." versus using actual syntax?
{noformat}
+ /**
+ * Verify that if the committer factory class is unknown, you cannot
+ * create committers.
+ */
+ @Test
+ public void testCommitterNullOutputPath() throws Throwable {
{noformat}
Update comment: "Verify that if the output path is null..."
{noformat}
+ /**
+ * Number of times to retry any repeatable S3 client request on failure,
+ * excluding throttling requests: {@value}.
+ */
+ public static final String RETRY_LIMIT = "fs.s3a.retry.limit";
{noformat}
Do we want to explain the difference from fs.s3a.attempts.maximum somewhere to
avoid confusion?
{noformat}
+ public <T> T retryUntranslated(
...
+ try {
+ // execute the operation, returning if successful
+ return operation.execute();
+ } catch (IOException | SdkBaseException e) {
+ caught = e;
+ }
+ // you only get here if the operation didn't complete
+ // normally, hence caught != null
{noformat}
Unless exception e is not the caught type. What if SDK bug yields NPE, for
example? Then you NPE in your code in translateException() guts, no?
{noformat}
+public class Retries {
{noformat}
Retries stuff great to see, but the change is invasive and needs separate
careful consideration IMO. How painful would this be to pull out into separate
patch?
First question here is, what about SDK retries? I know the SDK documentation
and behavior is spotty at best, but there are some known things (i.e. DynamoDB
does retry in SDK for most operations, exclusive of batched operations which
are application responsibility).
{noformat}
+
+ // does the output stream have delayed visibility
+ case CommitConstants.STREAM_CAPABILITY_MAGIC_OUTPUT:
+ return !putTracker.outputImmediatelyVisible();
{noformat}
StreamCapabilites is a nice addition and works well here IMO.
{noformat}
+ * This declared delete as idempotent.
+ * This is an "interesting" topic in past Hadoop FS work.
+ * Essentially: with a single caller, DELETE is idempotent
+ * but in a shared filesystem, it is is very much not so.
+ * Here, on the basis that isn't a filesystem with consistency guarantees,
+ * retryable results in files being deleted.
+ */
+ private static final boolean DELETE_CONSIDERED_IDEMPOTENT = true;
{noformat}
Yes, interesting. Is the argument that delete is racy anyways, even if clients
externally synchronize, thus we treat it as idempotent even though it is not?
Anyways, retry on failure can lead to success, but retry after (a) a request
that did end up deleting it or (b) after a different client deleted from under
us will cause an error after retries are exhausted (since subsequent retries
will always fail to delete). Do we want a different retry policy for this sort
of thing that is not-really-idempotent-but-we-retry-anyways?
{noformat}
+ abortOutstandingMultipartUploads(purgeExistingMultipartAge);
+ } catch (AccessDeniedException e) {
+ instrumentation.errorIgnored();
+ LOG.debug("Failed to purging multipart uploads against {}," +
+ " FS may be read only", bucket, e);
{noformat}
LOG.info()? I feel like folks might want to know their purge policy is not
working. You can always shut it up with per-bucket config.
{noformat}
+ @Retries.RetryRaw
{noformat}
I like these annotations. Pretty much necessary to reason about the surface
area / scope around retries and translations. Now if we could get the same for
the SDK, please. ;-)
{noformat}
+ * Although the committer APIs allow for a committer to be created without
+ * an output path, this is no supported in this class or its subclasses:
{noformat}
/no/not/
{noformat}
+public final class Tasks {
{noformat}
Didn't run this through the wet compiler carefully yet. Are there unit tests
somewhere I didn't notice?
{noformat}
+ try (DurationInfo d = new DurationInfo("%s: aborting job in state %s ",
+ r, CommitUtils.jobIdString(context), state)) {
{noformat}
How about passing a Logger into DurationInfo instead of it using its own?
{noformat}
+ public void cleanupStagingDirs() throws IOException {
+
+ }
{noformat}
{noformat}
+ protected void cleanup(JobContext context, boolean suppressExceptions)
+ throws IOException {
+
+ }
{noformat}
Just curious, could these be abstract, or do you want a default no-op without
making it explicit in subclass?
{noformat}
+ throws IOException {
+ if (pending == null || pending.isEmpty()) {
+ LOG.info("{}: no pending commits to abort", getRole());
+ } else {
+ Tasks.foreach(pending)
+ .throwFailureWhenFinished(!suppressExceptions)
+ .executeWith(buildThreadPool(context))
+ .onFailure((commit, exception) ->
+ getCommitOperations().abortSingleCommit(commit))
+ .run(commit -> getCommitOperations().abortSingleCommit(commit));
{noformat}
Failure callback is same as intended operation. Is this in effect a single
retry?
{noformat}
+/**
+ * Dynamically create the output committer based on the filesystem type.
+ * For S3A output, uses the committer provided in the implementation class.
+ * For other filesystems, returns the classic committer.
{noformat}
Does it? Looks like it throws a "FS not supported" exception.
{noformat}
+ private void innerCommit(SinglePendingCommit commit) throws IOException {
+ // finalize the commit
+ writeOperations.completeMPUwithRetries(
+ commit.getDestinationKey(),
+ commit.getUploadId(),
+ CommitUtils.toPartEtags(commit.getEtags()),
+ commit.getLength(),
+ new AtomicInteger(0));
{noformat}
Curious: why an atomic here?
{noformat}
+ /**
+ * Robust list files.
+ * @param path path
+ * @param recursive recursive listing?
+ * @return iterator (which is *not* robust)
{noformat}
Why robust? Did you used to have retry logic here? Now it just calls
underlying function. Same with the function below it.
{noformat}
+public class DynamicCommitterFactory extends AbstractS3GuardCommitterFactory {
{noformat}
General naming comment: Should these be called S3A Committers instead of
S3Guard Committers, now that one of the main implementations is orthogonal to
S3Guard? Same comment goes for the docs.
{noformat}
+/**
+ * Path operations for the staging committers.
+ */
+public final class Paths {
{noformat}
Can you comment on test coverage for this class? I didn't see a unit test but
may have missed it.
{noformat}
- res = dynamoDB.batchWriteItemUnprocessed(unprocessed);
- unprocessed = res.getUnprocessedItems();
+ final Map<String, List<WriteRequest>> p = unprocessed;
+ unprocessed = dataAccess.retry("batchWrite", "", true,
+ () ->
+ dynamoDB.batchWriteItemUnprocessed(p).getUnprocessedItems()
+ );
{noformat}
I think you want to remove the retry here. The existing function appears to
work correctly for retries
([HADOOP-13904|https://issues.apache.org/jira/browse/HADOOP-13904]) as they are
"supposed" to work. You are retrying the retries inside of an existing retry
loop but not retrying the original operation above it. I found it confusing.
In general, especially in this dynamo class, it seems like we're covering
things that are supposed to be retried by SDK with retries. I'm not sure the
benefit of that versus the risk of breaking something or eliminating fail-fast
behavior when it is warranted?
> Add S3Guard committer for zero-rename commits to S3 endpoints
> -------------------------------------------------------------
>
> Key: HADOOP-13786
> URL: https://issues.apache.org/jira/browse/HADOOP-13786
> Project: Hadoop Common
> Issue Type: New Feature
> Components: fs/s3
> Affects Versions: 3.0.0-beta1
> Reporter: Steve Loughran
> Assignee: Steve Loughran
> Attachments: HADOOP-13786-036.patch, HADOOP-13786-037.patch,
> HADOOP-13786-038.patch, HADOOP-13786-039.patch,
> HADOOP-13786-HADOOP-13345-001.patch, HADOOP-13786-HADOOP-13345-002.patch,
> HADOOP-13786-HADOOP-13345-003.patch, HADOOP-13786-HADOOP-13345-004.patch,
> HADOOP-13786-HADOOP-13345-005.patch, HADOOP-13786-HADOOP-13345-006.patch,
> HADOOP-13786-HADOOP-13345-006.patch, HADOOP-13786-HADOOP-13345-007.patch,
> HADOOP-13786-HADOOP-13345-009.patch, HADOOP-13786-HADOOP-13345-010.patch,
> HADOOP-13786-HADOOP-13345-011.patch, HADOOP-13786-HADOOP-13345-012.patch,
> HADOOP-13786-HADOOP-13345-013.patch, HADOOP-13786-HADOOP-13345-015.patch,
> HADOOP-13786-HADOOP-13345-016.patch, HADOOP-13786-HADOOP-13345-017.patch,
> HADOOP-13786-HADOOP-13345-018.patch, HADOOP-13786-HADOOP-13345-019.patch,
> HADOOP-13786-HADOOP-13345-020.patch, HADOOP-13786-HADOOP-13345-021.patch,
> HADOOP-13786-HADOOP-13345-022.patch, HADOOP-13786-HADOOP-13345-023.patch,
> HADOOP-13786-HADOOP-13345-024.patch, HADOOP-13786-HADOOP-13345-025.patch,
> HADOOP-13786-HADOOP-13345-026.patch, HADOOP-13786-HADOOP-13345-027.patch,
> HADOOP-13786-HADOOP-13345-028.patch, HADOOP-13786-HADOOP-13345-028.patch,
> HADOOP-13786-HADOOP-13345-029.patch, HADOOP-13786-HADOOP-13345-030.patch,
> HADOOP-13786-HADOOP-13345-031.patch, HADOOP-13786-HADOOP-13345-032.patch,
> HADOOP-13786-HADOOP-13345-033.patch, HADOOP-13786-HADOOP-13345-035.patch,
> cloud-intergration-test-failure.log, objectstore.pdf, s3committer-master.zip
>
>
> A goal of this code is "support O(1) commits to S3 repositories in the
> presence of failures". Implement it, including whatever is needed to
> demonstrate the correctness of the algorithm. (that is, assuming that s3guard
> provides a consistent view of the presence/absence of blobs, show that we can
> commit directly).
> I consider ourselves free to expose the blobstore-ness of the s3 output
> streams (ie. not visible until the close()), if we need to use that to allow
> us to abort commit operations.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
