Merge pull request #983: [proxima-beam-core] fix packages of retract transforms

Author: je-ik

beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/KeyedRetractPCollection.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
 
 import cz.o2.proxima.core.functional.UnaryFunction;
 import cz.o2.proxima.core.util.ExceptionUtils;

beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/LeftOrRight.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
 
 import java.io.IOException;
 import java.io.InputStream;

beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/RetractElement.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
 
 import cz.o2.proxima.core.functional.UnaryFunction;
 import java.io.IOException;

beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/RetractInnerJoin.java

@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
 
-import cz.o2.proxima.beam.core.transform.retract.LeftOrRight.LeftOrRightCoder;
+import cz.o2.proxima.beam.core.transforms.retract.LeftOrRight.LeftOrRightCoder;
 import cz.o2.proxima.beam.util.state.ExcludeExternal;
 import cz.o2.proxima.core.functional.UnaryFunction;
 import cz.o2.proxima.core.util.ExceptionUtils;
@@ -54,7 +54,7 @@
 import org.joda.time.Instant;
 
 @Slf4j
-public class RetractMostRecentJoin {
+public class RetractJoin {
 
   public static <K1, K2, V1, V2, JK> RetractPCollection<KV<KV<K1, K2>, KV<V1, V2>>> join(
       KeyedRetractPCollection<K1, V1> lhs,
@@ -206,9 +206,10 @@ public void process(
 
       Instant cleanupStamp = cleanupTs.read();
       if (cleanupStamp == null) {
-        // first element setup timer
+        // first element set up timer
         cleanupTimer.offset(cleanupDuration).setRelative();
-        cleanupTs.write(cleanupStamp = BoundedWindow.TIMESTAMP_MIN_VALUE);
+        cleanupStamp = BoundedWindow.TIMESTAMP_MIN_VALUE;
+        cleanupTs.write(cleanupStamp);
       }
 
       AtomicLong seq = new AtomicLong(MoreObjects.firstNonNull(seqState.read(), 0L));
@@ -257,8 +258,8 @@ private <PK, SK, PV, SV> void retractAndEmit(
         PK primaryKey,
         RetractElement<PV> primaryValue,
         Instant ts,
-        OutputConsumer<PK, SK, PV, SV> add,
-        OutputConsumer<PK, SK, PV, SV> retract) {
+        OutputConsumer<PK, SK, PV, SV> addConsumer,
+        OutputConsumer<PK, SK, PV, SV> retractConsumer) {
 
       SequentialInstant thisInstant = new SequentialInstant(ts, primaryValue.getSeqId());
       @Nullable KV<SequentialInstant, RetractElement<PV>> oldPrimaryValue =
@@ -288,36 +289,55 @@ private <PK, SK, PV, SV> void retractAndEmit(
 
       // infer time-range for updates
       for (Map.Entry<SK, KV<SequentialInstant, RetractElement<SV>>> e : secondaryKeyedElements) {
-        SK secondaryKey = e.getKey();
-        KV<SequentialInstant, RetractElement<SV>> secondaryValue = e.getValue();
-        // if secondary value exists
-        if (secondaryValue.getValue().isAddition()) {
-          // remove old value, if exists
-          if (oldPrimaryValue != null && oldPrimaryValue.getValue().isAddition()) {
-            retract.apply(
-                primaryKey,
-                secondaryKey,
-                oldPrimaryValue.getValue().getValue(),
-                secondaryValue.getValue().getValue(),
-                max(oldPrimaryValue.getKey(), secondaryValue.getKey()));
-          }
-          if (primaryValue.isAddition()) {
-            // insert new value
-            add.apply(
-                primaryKey,
-                secondaryKey,
-                primaryValue.getValue(),
-                secondaryValue.getValue().getValue(),
-                max(thisInstant, secondaryValue.getKey()));
-          } else {
-            // retract old value
-            retract.apply(
-                primaryKey,
-                secondaryKey,
-                oldPrimaryValue.getValue().getValue(),
-                secondaryValue.getValue().getValue(),
-                max(thisInstant, secondaryValue.getKey()));
-          }
+        processSecondaryElement(
+            primaryKey,
+            primaryValue,
+            e,
+            thisInstant,
+            oldPrimaryValue,
+            addConsumer,
+            retractConsumer);
+      }
+    }
+
+    private <PK, SK, PV, SV> void processSecondaryElement(
+        PK primaryKey,
+        RetractElement<PV> primaryValue,
+        Entry<SK, KV<SequentialInstant, RetractElement<SV>>> e,
+        SequentialInstant thisInstant,
+        @Nullable KV<SequentialInstant, RetractElement<PV>> oldPrimaryValue,
+        OutputConsumer<PK, SK, PV, SV> addConsumer,
+        OutputConsumer<PK, SK, PV, SV> retractConsumer) {
+
+      SK secondaryKey = e.getKey();
+      KV<SequentialInstant, RetractElement<SV>> secondaryValue = e.getValue();
+      // if secondary value exists
+      if (secondaryValue.getValue().isAddition()) {
+        // remove old value, if exists
+        if (oldPrimaryValue != null && oldPrimaryValue.getValue().isAddition()) {
+          retractConsumer.apply(
+              primaryKey,
+              secondaryKey,
+              oldPrimaryValue.getValue().getValue(),
+              secondaryValue.getValue().getValue(),
+              max(oldPrimaryValue.getKey(), secondaryValue.getKey()));
+        }
+        if (primaryValue.isAddition()) {
+          // insert new value
+          addConsumer.apply(
+              primaryKey,
+              secondaryKey,
+              primaryValue.getValue(),
+              secondaryValue.getValue().getValue(),
+              max(thisInstant, secondaryValue.getKey()));
+        } else {
+          // retract old value
+          retractConsumer.apply(
+              primaryKey,
+              secondaryKey,
+              oldPrimaryValue.getValue().getValue(),
+              secondaryValue.getValue().getValue(),
+              max(thisInstant, secondaryValue.getKey()));
         }
       }
     }

…forms/retract/RetractMostRecentJoin.java …core/transforms/retract/RetractJoin.javabeam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/RetractMostRecentJoin.java renamed to beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/RetractJoin.java beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/RetractMostRecentJoin.java renamed to beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/RetractJoin.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
 
 import cz.o2.proxima.core.functional.UnaryFunction;
 import cz.o2.proxima.core.util.ExceptionUtils;

beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/RetractPCollection.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
 
 import java.io.IOException;
 import java.io.InputStream;

beam/core/src/main/java/cz/o2/proxima/beam/core/transforms/retract/SequentialInstant.java

@@ -13,7 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
+
+import static org.junit.Assert.assertNotNull;
 
 import cz.o2.proxima.core.util.Pair;
 import java.util.ArrayList;
@@ -51,7 +53,7 @@
 import org.joda.time.Instant;
 import org.junit.Test;
 
-public class RetractMostRecentJoinTest {
+public class RetractJoinTest {
 
   final Duration cleanupDuration = Duration.standardDays(2);
 
@@ -71,7 +73,7 @@ public void testSimple() {
     PCollection<
             RetractElement<KV<KV<String, Integer>, KV<KV<String, Integer>, KV<Integer, String>>>>>
         joined =
-            RetractMostRecentJoin.join(
+            RetractJoin.join(
                     lhs, rhs, KV::getValue, KV::getKey, TypeDescriptors.integers(), cleanupDuration)
                 .unwrapped();
     PCollection<String> res =
@@ -89,7 +91,7 @@ public void testSimple() {
                             e.getValue().getValue().getValue().getValue())));
 
     PAssert.that(res).containsInAnyOrder("b:2::b:2:2:c", "a:1::a:1:1:b");
-    p.run();
+    assertNotNull(p.run());
   }
 
   @Test
@@ -124,7 +126,7 @@ public void testWithRetractions() {
     PCollection<
             RetractElement<KV<KV<String, Integer>, KV<KV<String, Integer>, KV<Integer, String>>>>>
         joined =
-            RetractMostRecentJoin.join(
+            RetractJoin.join(
                     lhs, rhs, KV::getValue, KV::getKey, TypeDescriptors.integers(), cleanupDuration)
                 .unwrapped();
     PCollection<String> res =
@@ -144,7 +146,7 @@ public void testWithRetractions() {
                                 e.getValue().getValue().getValue())));
 
     PAssert.that(res).containsInAnyOrder("a:3::a:3:3:c", "b:4::b:4:4:x");
-    p.run();
+    assertNotNull(p.run());
   }
 
   @Test
@@ -183,7 +185,7 @@ public void testWithManyRetractions() {
     PCollection<
             RetractElement<KV<KV<String, Integer>, KV<KV<String, Integer>, KV<Integer, String>>>>>
         joined =
-            RetractMostRecentJoin.join(
+            RetractJoin.join(
                     lhs, rhs, KV::getValue, KV::getKey, TypeDescriptors.integers(), cleanupDuration)
                 .unwrapped();
     PCollection<String> res =
@@ -203,7 +205,7 @@ public void testWithManyRetractions() {
                                 e.getValue().getValue().getValue())));
 
     PAssert.that(res).containsInAnyOrder(computeOutputs(inputs));
-    p.run();
+    assertNotNull(p.run());
   }
 
   private List<String> computeOutputs(

beam/core/src/test/java/cz/o2/proxima/beam/core/transforms/retract/RetractInnerJoinTest.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package cz.o2.proxima.beam.core.transform.retract;
+package cz.o2.proxima.beam.core.transforms.retract;
 
 import static org.junit.Assert.assertNotNull;