commit c08f39d9b46584d3a136e9f5782e09c481b7497a
author mynameborat <bharath.kumarasubramanian@gmail.com> Mon Mar 15 06:51:28 2021 -0700
committer GitHub <noreply@github.com> Mon Mar 15 06:51:28 2021 -0700
tree 0f33a60ddf17a8c58b9721e376a41ad1527e4117
parent 2aea5f71f9742bf995e88608750a071c44ceade5

[SAMZA-2597]: Maintain traversal order for registered operators (#1436)

samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java

diff --git a/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java b/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
index 9fd35eb..c8e2f7e 100644
--- a/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
+++ b/samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java
@@ -50,7 +50,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.Set;
 
 
@@ -109,9 +109,9 @@
     }
 
     this.highResClock = createHighResClock(context.getJobContext().getConfig());
-    registeredOperators = new HashSet<>();
-    prevOperators = new HashSet<>();
-    inputStreams = new HashSet<>();
+    registeredOperators = new LinkedHashSet<>();
+    prevOperators = new LinkedHashSet<>();
+    inputStreams = new LinkedHashSet<>();
 
     final ContainerContext containerContext = context.getContainerContext();
     final MetricsRegistry metricsRegistry = containerContext.getContainerMetricsRegistry();

samza-core/src/test/java/org/apache/samza/operators/impl/TestOperatorImpl.java

diff --git a/samza-core/src/test/java/org/apache/samza/operators/impl/TestOperatorImpl.java b/samza-core/src/test/java/org/apache/samza/operators/impl/TestOperatorImpl.java
index aea35aa..e86fadf 100644
--- a/samza-core/src/test/java/org/apache/samza/operators/impl/TestOperatorImpl.java
+++ b/samza-core/src/test/java/org/apache/samza/operators/impl/TestOperatorImpl.java
@@ -20,6 +20,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -39,6 +40,9 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyObject;
 import static org.mockito.Matchers.anyString;
@@ -79,6 +83,31 @@
   }
 
   @Test
+  public void testRegisterOperatorMaintainsInsertionOrder() {
+    OperatorImpl<Object, Object> opImpl = new TestOpImpl(mock(Object.class));
+    opImpl.init(this.internalTaskContext);
+
+    OperatorImpl<Object, Object> firstOp = mock(OperatorImpl.class);
+    OperatorImpl<Object, Object> secondOp = mock(OperatorImpl.class);
+    OperatorImpl<Object, Object> thirdOp = mock(OperatorImpl.class);
+
+    opImpl.registerNextOperator(firstOp);
+    opImpl.registerNextOperator(secondOp);
+    opImpl.registerNextOperator(thirdOp);
+
+    Iterator<OperatorImpl<Object, ?>> iterator = opImpl.registeredOperators.iterator();
+    assertTrue("Expecting non-empty iterator", iterator.hasNext());
+    assertEquals("Expecting first operator to be returned", firstOp, iterator.next());
+
+    assertTrue("Expecting non-empty iterator", iterator.hasNext());
+    assertEquals("Expecting second operator to be returned", secondOp, iterator.next());
+
+    assertTrue("Expecting non-empty iterator", iterator.hasNext());
+    assertEquals("Expecting third operator to be returned", thirdOp, iterator.next());
+
+    assertFalse("Expecting no more registered operators", iterator.hasNext());
+  }
+  @Test
   public void testOnMessagePropagatesResults() {
     Object mockTestOpImplOutput = mock(Object.class);
     OperatorImpl<Object, Object> opImpl = new TestOpImpl(mockTestOpImplOutput);