[BEAM-10141] HL7v2 io timestamps / watermark estimate #11862

jaketf · 2020-05-29T17:15:34Z

Use timestamps / watermark estimate in HL7v2IO.[Read, ListHL7v2Messages] and provide interface to opt-in to TimestampedValue<HL7v2Message> to preserve the existing interface.

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.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza
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.

lukecwik · 2020-06-02T17:53:34Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

@@ -179,25 +189,51 @@ public static ListHL7v2Messages readAll(List<String> hl7v2Stores) {
    return new ListHL7v2Messages(StaticValueProvider.of(hl7v2Stores), StaticValueProvider.of(null));
  }

+  /** Read all HL7v2 Messages from multiple stores as sendTime {@link TimestampedValue}s. */
+  public static ListTimestampedHL7v2Messages readAllWithTimestamps(List<String> hl7v2Stores) {


Instead of adding static methods for each variant, any reason to not make ListHL7v2Messages be a builder so someone could invoke withMessageTimestamps?

This would have also make sense to make withFilter a builder like method on ListHL7v2Messages

this is a good point.

Unfortunately, I think having adding replacing the existing static methods with withFilter builder would be interface breaking. I've added this method to ListHL7v2Messages as it might be more natural for some to use.

I think you can keep the static methods that exist and transition them to use the builder underneath so you don't have to break the interface. This is done in other IOs for common scenarios.

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

aaltay · 2020-06-12T00:01:14Z

retest this please

aaltay · 2020-06-18T19:23:57Z

retest this please

aaltay · 2020-06-18T19:26:28Z

retest this please

lukecwik · 2020-06-18T23:04:41Z

Sorry for the long delay.

lukecwik · 2020-06-18T23:06:15Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

+     * @param initialSplitDuration the initial split duration
+     * @return the list hl 7 v 2 messages
+     */
+    ListHL7v2Messages withInitialSplitDuration(Duration initialSplitDuration) {
      this.initialSplitDuration = initialSplitDuration;


Please return a new version of this transform instead of mutating the existing version.

It is not uncommon for users to take one transform apply it to the graph once and then call withX on it and apply it elsewhere. We wouldn't want the second mutation to ever impact the first application of the transform and it is easy to guard against it using the @autovalue builders.

lukecwik · 2020-06-18T23:06:32Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

+     * disjoint sendTime filters of the specified duration.
+     *
+     * @param initialSplitDuration the initial split duration
+     * @return the list hl 7 v 2 messages


lukecwik · 2020-06-18T23:08:33Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

+     * or the input element's timestamp.
+     *
+     * @param timestampMethod the timestamp method
+     * @return the list hl 7 v 2 messages


lukecwik · 2020-06-18T23:08:52Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

+     * Controls if the output elements will be assigned a timestamp beased on the sendTime property
+     * or the input element's timestamp.


Suggested change

* Controls if the output elements will be assigned a timestamp beased on the sendTime property

* or the input element's timestamp.

* Controls if the output elements will be assigned a timestamp beased on the {@code sendTime} property

* or the input element's timestamp.

lukecwik · 2020-06-18T23:10:30Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

+     * @param timestampMethod the timestamp method
+     * @return the list hl 7 v 2 messages
+     */
+    ListHL7v2Messages withTimestampMethod(TimestampMethod timestampMethod) {


How about withOutputTimestampMethod, also check that the argument is non-null?

nit: I'm torn between using an enum or just having the two variants between methods. Your call. Having two variants prevents the user from passing in null here.

lukecwik · 2020-06-18T23:11:41Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

- * <p>Message Listing Message Listing with {@link HL7v2IO.ListHL7v2Messages} supports batch use
- * cases where you want to process all the messages in an HL7v2 store or those matching a
- * filter @see <a
+ * <p>Message Listing Message Listing with {@link ListHL7v2Messages} and {@link ListHL7v2Messages}


?

Suggested change

* <p>Message Listing Message Listing with {@link ListHL7v2Messages} and {@link ListHL7v2Messages}

* <p>Message Listing with {@link ListHL7v2Messages} and {@link ListHL7v2Messages}

lukecwik · 2020-06-18T23:21:00Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

+            outputReceiver.outputWithTimestamp(msg, Instant.parse(msg.getSendTime()));
+            break;
+          case INPUT_ELEMENT:
+            outputReceiver.outputWithTimestamp(msg, context.timestamp());


You should use output(msg) and not outputWithTimestamp since output() by default does this. This also allows you to not have to take a ProcessContext parameter.

lukecwik · 2020-06-18T23:22:32Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

@@ -509,13 +561,17 @@ public ListHL7v2Messages withInitialSplitDuration(Duration initialSplitDuration)
     * @param filter the filter
     */
    ListHL7v2MessagesFn(String filter) {
-      this(StaticValueProvider.of(filter), null);
+      this(StaticValueProvider.of(filter), null, null);


We pass in null to the constructor and don't handle that case in the switch statement. How about passing in the default?

lukecwik · 2020-06-18T23:24:23Z

...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java

    private Duration initialSplitDuration;
+    private TimestampMethod timestampMethod;
+
+    enum TimestampMethod {


lukecwik · 2020-08-21T22:20:09Z

Ping?

aaltay · 2020-09-01T19:13:11Z

@jaketf - Is this PR still active?

jaketf · 2020-09-01T19:24:03Z

apologies I'm not working w/ healthcare clients these days and unable to prioritize this.
I can take a look at this again after my current engagements end (~october)

stale · 2020-11-07T21:59:00Z

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.

stale · 2020-11-16T03:39:04Z

This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

Jacob Ferriero added 5 commits May 28, 2020 13:44

[BEAM-10141] use timestamps / watermark estimate in HL7v2IO ListMessages

3d1bb3b

add option for sendTime timestamps to HL7v2IO.Read

992fa2f

fix typo add tests

6bdc6b5

update CHANGES.md

54888b1

add docs note about input PCollections and timestamps for HL7v2IO.Read

388d07c

probot-autolabeler bot added gcp io java labels May 29, 2020

iemejia requested a review from lukecwik May 30, 2020 14:59

fix tests

08ce4cd

lukecwik requested changes Jun 2, 2020

View reviewed changes

spotless

970c063

jaketf requested a review from lukecwik June 3, 2020 02:48

Merge branch 'master' into HL7v2IOTimestamps

0176ab1

lukecwik requested changes Jun 18, 2020

View reviewed changes

jaketf marked this pull request as draft September 1, 2020 19:23

stale bot added the stale label Nov 7, 2020

stale bot closed this Nov 16, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-10141] HL7v2 io timestamps / watermark estimate #11862

[BEAM-10141] HL7v2 io timestamps / watermark estimate #11862

jaketf commented May 29, 2020

lukecwik Jun 2, 2020

jaketf Jun 3, 2020

lukecwik Jun 18, 2020

aaltay commented Jun 12, 2020

aaltay commented Jun 18, 2020

aaltay commented Jun 18, 2020

lukecwik commented Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik Jun 18, 2020

lukecwik commented Aug 21, 2020

aaltay commented Sep 1, 2020

jaketf commented Sep 1, 2020

stale bot commented Nov 7, 2020

stale bot commented Nov 16, 2020

		* Controls if the output elements will be assigned a timestamp beased on the sendTime property
		* or the input element's timestamp.

	* <p>Message Listing Message Listing with {@link ListHL7v2Messages} and {@link ListHL7v2Messages}
	* <p>Message Listing with {@link ListHL7v2Messages} and {@link ListHL7v2Messages}

[BEAM-10141] HL7v2 io timestamps / watermark estimate #11862

[BEAM-10141] HL7v2 io timestamps / watermark estimate #11862

Conversation

jaketf commented May 29, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aaltay commented Jun 12, 2020

aaltay commented Jun 18, 2020

aaltay commented Jun 18, 2020

lukecwik commented Jun 18, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik commented Aug 21, 2020

aaltay commented Sep 1, 2020

jaketf commented Sep 1, 2020

stale bot commented Nov 7, 2020

stale bot commented Nov 16, 2020