From bdc738e0c010829af7f7649f6ea57dc044a18285 Mon Sep 17 00:00:00 2001 From: Daniel Mills Date: Tue, 12 Sep 2017 16:54:36 -0700 Subject: [PATCH 1/3] Mark AuctionOrBidWindowCoder as consistentWithEquals --- .../org/apache/beam/sdk/nexmark/queries/WinningBids.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java index 816a81f0d0a9..efdbe21ba9f3 100644 --- a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java +++ b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java @@ -192,6 +192,11 @@ public AuctionOrBidWindow decode(InputStream inStream) } @Override public void verifyDeterministic() throws NonDeterministicException {} + + @Override + public boolean consistentWithEquals() { + return true; + } } /** Assign events to auction windows and merges them intelligently. */ From e93c035711290d8c8ccefdae2e314faef7164179 Mon Sep 17 00:00:00 2001 From: Daniel Mills Date: Wed, 13 Sep 2017 10:08:22 -0700 Subject: [PATCH 2/3] Enforce correctness of consistentWithEquals and fix broken hashCode() method for AuctionOrBidWindowCoder --- .../org/apache/beam/sdk/nexmark/queries/WinningBids.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java index efdbe21ba9f3..be649582bad3 100644 --- a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java +++ b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java @@ -155,7 +155,7 @@ public String toString() { } @Override public int hashCode() { - return Objects.hash(isAuctionWindow, auction); + return Objects.hash(super.hashCode(), isAuctionWindow, auction); } } @@ -167,6 +167,8 @@ private static class AuctionOrBidWindowCoder extends CustomCoder SUPER_CODER = IntervalWindow.getCoder(); private static final Coder ID_CODER = VarLongCoder.of(); private static final Coder INT_CODER = VarIntCoder.of(); + private static final boolean CONSISTENT_WITH_EQUALS = SUPER_CODER.consistentWithEquals() + && ID_CODER.consistentWithEquals() && INT_CODER.consistentWithEquals(); @JsonCreator public static AuctionOrBidWindowCoder of() { @@ -195,7 +197,7 @@ public AuctionOrBidWindow decode(InputStream inStream) @Override public boolean consistentWithEquals() { - return true; + return CONSISTENT_WITH_EQUALS; } } From e3abd6988b9e9bf0ad465dda573870650ddceb2e Mon Sep 17 00:00:00 2001 From: Daniel Mills Date: Wed, 13 Sep 2017 10:21:01 -0700 Subject: [PATCH 3/3] Make AuctionOrBidWindowCoder use structuralValue instead of consistentWithEquals --- .../org/apache/beam/sdk/nexmark/queries/WinningBids.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java index be649582bad3..d73b8ae69541 100644 --- a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java +++ b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/queries/WinningBids.java @@ -167,8 +167,6 @@ private static class AuctionOrBidWindowCoder extends CustomCoder SUPER_CODER = IntervalWindow.getCoder(); private static final Coder ID_CODER = VarLongCoder.of(); private static final Coder INT_CODER = VarIntCoder.of(); - private static final boolean CONSISTENT_WITH_EQUALS = SUPER_CODER.consistentWithEquals() - && ID_CODER.consistentWithEquals() && INT_CODER.consistentWithEquals(); @JsonCreator public static AuctionOrBidWindowCoder of() { @@ -196,8 +194,8 @@ public AuctionOrBidWindow decode(InputStream inStream) @Override public void verifyDeterministic() throws NonDeterministicException {} @Override - public boolean consistentWithEquals() { - return CONSISTENT_WITH_EQUALS; + public Object structuralValue(AuctionOrBidWindow value) { + return value; } }