[BEAM-8374] Enable returning missing PublishResult fields in SnsIO.Write

[jfarr]

This PR adds new coders for returning the entire PublishResult object from SnsIO, or the entire PublishResult without HTTP headers. This is necessary in order to check the HTTP status for errors since the current implementation only encodes the messageId field. These new coders can be selected by calling the new withFullPublishResult() and withFullPublishResultWithoutHeaders() methods on SnsIO.Write. The withCoder() method also allows setting a custom coder. This PR also adds coders for common AWS SDK objects ResponseMetadata and SdkHttpMetadata.
R: @lukecwik

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---		---	---
Java
Python		---			---	---
XLang	---	---	---		---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

[kennknowles]

@jhalaria @iemejia

[lukecwik]

Note that this is a breaking change to any Dataflow streaming pipelines using the PublishResult.

[jfarr]

Note that this is a breaking change to any Dataflow streaming pipelines using the PublishResult.

@lukecwik Thanks for the feedback. I did not realize that, I'm very new to Beam. What do you suggest would be the best way to deal with that?

[iemejia]

Last week we merged SnsIO based on Amazon SDK for Java v2 I was wondering if these fixes could apply there too. We may consider this optional for this PR but maybe worth for consistency to check if we can have a similar approach there too. CC @cmachgodaddy

[iemejia]

One extra issue that I have not really thought about is the size difference that we will have by adding the full headers/sdk metadata to all answers, mmm... now I am hesitating about maybe having a V2 like approach for the cases where we care about the full metadata (akin to FileIO).
What is the use case that is pushing into this approach?

[jfarr]

One extra issue that I have not really thought about is the size difference that we will have by adding the full headers/sdk metadata to all answers, mmm... now I am hesitating about maybe having a V2 like approach for the cases where we care about the full metadata (akin to FileIO).
What is the use case that is pushing into this approach?

@iemejia for our use case we only need the HTTP status. We have written an SNS sink that uses this to check for error responses. I just included HTTP headers for completeness but I can remove them if it's a concern. In that case I wonder if a separate coder for SdkHttpMetadata makes sense though.

[jfarr]

Last week we merged SnsIO based on Amazon SDK for Java v2 I was wondering if these fixes could apply there too. We may consider this optional for this PR but maybe worth for consistency to check if we can have a similar approach there too. CC @cmachgodaddy

Sure, I work with Cam and we discussed it today. I can apply the changes to v2 in this PR, no problem.

[cmachgodaddy]

I wonder if we should add unit test for this coder? I know we did not have it, b/c we had just one property to encode and decode. But, now we have quite a few.

[jfarr]

@iemejia @lukecwik How does this look? If you're happy with it I'll port it over to amazon-web-services2.

[lukecwik]

I would suggest keeping the "existing" coder and allowing users to choose a coder that encodes all the additional metadata and http response code. This way users who need the additional details could use a builder style method within the IO class like "withMetadata()" or "withFullMessage" which ensures that the full message is encoded. The class comment could say by default we only provide fields X and Y but if you need everything then specify "with..."

By making this change you won't break existing users and it won't impact their performance and it will enable your usecase. Finally for the V2 version we may want to use the "full" coder by default and use the stripped down coder when told via a "withMessageIdOnly()" or something along those lines.

[jfarr]

@lukecwik @iemejia I reverted the original coder and added a new "full" (all fields) and "minimal" (no HTTP headers) coder, and also factored out new coders for the common SDK objects. If this looks more like it then I'll flesh out the unit tests, please let me know.

[lukecwik]

The direction looks good, will let you follow up on the comments and fill out the tests.

[jfarr]

Hi @lukecwik thank you for the feedback. I believe all of the requested changes have been addressed now.

[stale]

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

[jfarr]

Hi @lukecwik @iemejia this PR was marked stale. I believe I have addressed all of your feedback. Is there anything else you would like me to change?

[lukecwik]

Sorry for the delay, taking a look now.

[lukecwik]

This looks great, mostly minor feedback.

[jfarr]

Hi @lukecwik and @iemejia, I believe I have incorporated all of your review feedback. I also made a few changes to the interface that I hope will improve clarity:

• Instead of the confusingly named fullSdkHttpMetadata() and minimalSdkHttpMetadata() methods, the AwsCoders class now has a method sdkHttpMetadata() which creates a coder that by default does not include the response headers, and if you want to include the headers there is also a method sdkHttpMetadata(boolean includeHeaders). I'm on the fence about getting rid of the boolean parameter and just including a method like sdkHttpMetadataIncludingHeaders() so let me know if you have a preference.

• I changed the names of the SnsIO methods to withFullPublishResultNoHeaders and withFullPublishResultIncludingHeaders. The default is still to encode only the messageId for backward compatibility. I also added a method to specify your own coder for future extensibility.

I hope this PR is ready to merge now, please let me know if it needs any additional changes.

[lukecwik]

Thanks for returning back to this. Taking a look now.

[lukecwik]

I added my suggestions to name changes inline as responses to your comments.

[lukecwik]

Run Java PreCommit

[lukecwik]

Run Java PreCommit

[jfarr]

Thanks for the feedback @lukecwik. I've addressed your comments and also added one more unit test in SnsIOTest. Please let me know what you think now.

[lukecwik]

retest this please

[jfarr]

@lukecwik thanks for the suggestions!

[jfarr]

One last question @lukecwik, would you prefer to apply these changes in amazon-web-services2 in this PR? Or should we take that to a separate PR? That might be a good opportunity to discuss making some improvements to how errors are handled in this IO. In my opinion it's somewhat suboptimal that IOExceptions and the like are handled in the IO but HTTP error responses have to be handled in the caller, for example.

[cmachgodaddy]

Thanks for the changes!

[cmachgodaddy]

Just curious, why this coder is not following the pattern coder.of(xxx) ?

[jfarr]

@cmachgodaddy That's because this class is designed as a factory for coders, with the coder implementation being an internal detail, as @lukecwik suggested in an earlier comment thread.

Style note, you can create one class called AwsCoders which has three static methods minimalSdkHttpMetadata() and fullSdkHttpMetadata() and responseMetadata() that return the coders. This way you can make all the coder implementations private inner static classes and they all become implementation details without needing to expose them to users. Users can still get an instance if they need one but it gives us flexibility from a maintenance point of view how the coders are organized.

I could create a static factory method of() on SdkHttpMetadata but it wouldn't serve any purpose, it would only be called by sdkHttpMetadata() which is already a static factory method. SdkHttpMetadata is not meant to be instantiated directly by the caller.

[lukecwik]

of() prevents moving the class while a top level static method allows for stuff to still be moved around.

[cmachgodaddy]

And one more curiosity, why we have to implement verifyDeterministic for this coder?

[lukecwik]

By default coders are considered non-deterministic which prevent them from being used as the key coder for many things (GBK, map/multimap side input keys, map state keys, ...).

Implementing it allows us to be deterministic if the coders we use are deterministic. So if the internally used coders become deterministic then we get that for free.

[jfarr]

CustomCoder's implementation always throws NonDeterministicException so this seemed more correct.

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CustomCoder.java

Lines 47 to 53 in 33ec0bb

	@Override
	public void verifyDeterministic() throws NonDeterministicException {
	throw new NonDeterministicException(
	this,
	"CustomCoder implementations must override verifyDeterministic,"
	+ " or they are presumed nondeterministic.");
	}

[cmachgodaddy]

Same here, any reasons this is not XXXcoder.of(...) pattern?

[jfarr]

It's sort of the same reason as above, PublishResultCoders is a factory for creating several different coder implementations so the factory methods have different names to indicate which one. I could rename this method to of() but since we are introducing new behavior I wanted to make it very clear that this is the default, or existing, behavior. Also I'm not sure if it makes sense to have a method called of() on the factory not on the coder itself.

[lukecwik]

One last question @lukecwik, would you prefer to apply these changes in amazon-web-services2 in this PR? Or should we take that to a separate PR? That might be a good opportunity to discuss making some improvements to how errors are handled in this IO. In my opinion it's somewhat suboptimal that IOExceptions and the like are handled in the IO but HTTP error responses have to be handled in the caller, for example.

I would prefer a separate PR so that people can start using this sooner then later. Also smaller PRs are easier to review.

[lukecwik]

@iemejia Any final comments?

Otherwise I intend to merge when the tests are green.

[lukecwik]

Run Java PreCommit

[lukecwik]

Run Java PreCommit