Allow SqlPtransform to run in streaming mode.

linkedin · Feb 8, 2024 · cccac3a · cccac3a
1 parent 0145270
commit cccac3a
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 6 deletions.
diff --git a/...rc/main/java/org/apache/beam/runners/flink/transform/sql/FlinkSQLTransformTranslator.java b/...rc/main/java/org/apache/beam/runners/flink/transform/sql/FlinkSQLTransformTranslator.java
@@ -58,10 +58,6 @@ class FlinkSQLTransformTranslator<InputT, OutputT>
   public void translateNode(
       PTransform<PCollection<InputT>, PCollection<OutputT>> transform,
       FlinkStreamingTranslationContext context) {
-    if (context.isStreaming()) {
-      throw new IllegalStateException(
-          "The current job is a streaming job. Flink SQL transform only support batch jobs.");
-    }
     MultiOutputSqlTransformWithInput<InputT, OutputT> sqlTransform =
         (MultiOutputSqlTransformWithInput) transform;
     StreamTableEnvironment tEnv = StreamTableEnvironment.create(context.getExecutionEnvironment());

diff --git a/...link/1.15/src/test/java/org/apache/beam/runners/flink/transform/sql/SqlTransformTest.java b/...link/1.15/src/test/java/org/apache/beam/runners/flink/transform/sql/SqlTransformTest.java
@@ -278,15 +278,17 @@ public void testOnlySetAdditionalInputForMultiOutputSqlTransform() {
     pipeline.run(getPipelineOptions());
   }
 
-  @Test(expected = IllegalStateException.class)
-  public void testApplySqlToStreamingJobThrowException() {
+  @Test
+  public void testStreamingMode() throws IOException {
     Pipeline pipeline = Pipeline.create();
     SingleOutputSqlTransform<FlinkSqlTestUtils.Order> transform =
         SqlTransform.of(FlinkSqlTestUtils.Order.class)
             .withDDL(ORDERS_DDL)
             .withQuery("SELECT orderNumber, product, amount, price, buyer, orderTime FROM Orders");
     pipeline.apply(transform);
 
+    PCollection<FlinkSqlTestUtils.Order> outputs = pipeline.apply(transform);
+    verifyRecords(outputs, "Orders", FlinkSqlTestUtils.Order.class);
     FlinkPipelineOptions options = getPipelineOptions();
     options.setStreaming(true);
     pipeline.run(options);