Extract getSourceParallelism() helper method

Extract the duplicated parallelism calculation logic from translateUnboundedSource
and translateBoundedSource into a shared private helper method.

Author: dnamaz

runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java

@@ -192,6 +192,12 @@ public static String getCurrentTransformName(FlinkStreamingTranslationContext co
     return context.getCurrentTransform().getFullName();
   }
 
+  /** Returns the parallelism to use for source operators. */
+  private static int getSourceParallelism(FlinkStreamingTranslationContext context) {
+    int maxParallelism = context.getExecutionEnvironment().getMaxParallelism();
+    return maxParallelism > 0 ? maxParallelism : context.getExecutionEnvironment().getParallelism();
+  }
+
   // --------------------------------------------------------------------------------------------
   //  Transformation Implementations
   // --------------------------------------------------------------------------------------------
@@ -222,10 +228,7 @@ private static <T> void translateUnboundedSource(
 
     String fullName = getCurrentTransformName(context);
     try {
-      int parallelism =
-          context.getExecutionEnvironment().getMaxParallelism() > 0
-              ? context.getExecutionEnvironment().getMaxParallelism()
-              : context.getExecutionEnvironment().getParallelism();
+      int parallelism = getSourceParallelism(context);
 
       FlinkUnboundedSource<T> unboundedSource =
           FlinkSource.unbounded(
@@ -274,10 +277,7 @@ private static <T> void translateBoundedSource(
     TypeInformation<WindowedValue<T>> outputTypeInfo = context.getTypeInfo(output);
 
     String fullName = getCurrentTransformName(context);
-    int parallelism =
-        context.getExecutionEnvironment().getMaxParallelism() > 0
-            ? context.getExecutionEnvironment().getMaxParallelism()
-            : context.getExecutionEnvironment().getParallelism();
+    int parallelism = getSourceParallelism(context);
 
     FlinkBoundedSource<T> flinkBoundedSource =
         FlinkSource.bounded(