chore: enable reduce java sdk e2e test (#1454)

Signed-off-by: Keran Yang <[email protected]>
numaproj · Jan 12, 2024 · 645bcc5 · 645bcc5
1 parent 85e76c7
commit 645bcc5
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 38 deletions.
diff --git a/test/sdks-e2e/sdks_test.go b/test/sdks-e2e/sdks_test.go
@@ -19,6 +19,7 @@ limitations under the License.
 package sdks_e2e
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 	"sync"
@@ -86,43 +87,42 @@ func (s *SDKsSuite) TestMapStreamUDFunctionAndSink() {
 		VertexPodLogContains("java-udsink", "hello", PodLogCheckOptionWithContainer("udsink"), PodLogCheckOptionWithCount(4))
 }
 
-// TODO(session) fix after updating java sdk
-//func (s *SDKsSuite) TestReduceSDK() {
-//	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
-//	defer cancel()
-//	w := s.Given().Pipeline("@testdata/simple-keyed-reduce-pipeline.yaml").
-//		When().
-//		CreatePipelineAndWait()
-//	defer w.DeletePipelineAndWait()
-//	pipelineName := "even-odd-sum"
-//
-//	// wait for all the pods to come up
-//	w.Expect().VertexPodsRunning()
-//
-//	done := make(chan struct{})
-//	go func() {
-//		// publish messages to source vertex, with event time starting from 60000
-//		startTime := 60000
-//		for i := 0; true; i++ {
-//			select {
-//			case <-ctx.Done():
-//				return
-//			case <-done:
-//				return
-//			default:
-//				eventTime := strconv.Itoa(startTime + i*1000)
-//				w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte("1")).WithHeader("X-Numaflow-Event-Time", eventTime)).
-//					SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte("2")).WithHeader("X-Numaflow-Event-Time", eventTime)).
-//					SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte("3")).WithHeader("X-Numaflow-Event-Time", eventTime))
-//			}
-//		}
-//	}()
-//
-//	w.Expect().
-//		VertexPodLogContains("java-udsink", "120", PodLogCheckOptionWithContainer("udsink")).
-//		VertexPodLogContains("java-udsink", "240", PodLogCheckOptionWithContainer("udsink"))
-//	done <- struct{}{}
-//}
+func (s *SDKsSuite) TestReduceSDK() {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	w := s.Given().Pipeline("@testdata/simple-keyed-reduce-pipeline.yaml").
+		When().
+		CreatePipelineAndWait()
+	defer w.DeletePipelineAndWait()
+	pipelineName := "even-odd-sum"
+
+	// wait for all the pods to come up
+	w.Expect().VertexPodsRunning()
+
+	done := make(chan struct{})
+	go func() {
+		// publish messages to source vertex, with event time starting from 60000
+		startTime := 60000
+		for i := 0; true; i++ {
+			select {
+			case <-ctx.Done():
+				return
+			case <-done:
+				return
+			default:
+				eventTime := strconv.Itoa(startTime + i*1000)
+				w.SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte("1")).WithHeader("X-Numaflow-Event-Time", eventTime)).
+					SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte("2")).WithHeader("X-Numaflow-Event-Time", eventTime)).
+					SendMessageTo(pipelineName, "in", NewHttpPostRequest().WithBody([]byte("3")).WithHeader("X-Numaflow-Event-Time", eventTime))
+			}
+		}
+	}()
+
+	w.Expect().
+		VertexPodLogContains("java-udsink", "120", PodLogCheckOptionWithContainer("udsink")).
+		VertexPodLogContains("java-udsink", "240", PodLogCheckOptionWithContainer("udsink"))
+	done <- struct{}{}
+}
 
 func (s *SDKsSuite) TestSourceTransformer() {
 	var wg sync.WaitGroup

diff --git a/test/sdks-e2e/testdata/simple-keyed-reduce-pipeline.yaml b/test/sdks-e2e/testdata/simple-keyed-reduce-pipeline.yaml
@@ -19,7 +19,7 @@ spec:
       udf:
         container:
           # compute the sum, see https://github.com/numaproj/numaflow-java/tree/main/examples/src/main/java/io/numaproj/numaflow/examples/reduce/sum
-          image: quay.io/numaio/numaflow-java/reduce-sum:v0.6.0
+          image: quay.io/numaio/numaflow-java/reduce-sum:v0.6.1
         groupBy:
           window:
             fixed: