[Flink] Set default max input split size to -1 (disabled)

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

@@ -313,11 +313,11 @@ public interface FlinkPipelineOptions
 
   void setFlinkConfDir(String confDir);
 
-  @Description("Set the maximum size of input split when data is read from a filesystem.")
-  @Default.Long(128 * 1024 * 1024)
-  Long getFileInputSplitMaxSizeBytes();
+  @Description("Set the maximum size of input split when data is read from a filesystem. -1 implies no max size.")
+  @Default.Long(-1)
+  Long getFileInputSplitMaxSizeMB();
 
-  void setFileInputSplitMaxSizeBytes(Long inputSplitMaxSizeBytes);
+  void setFileInputSplitMaxSizeMB(Long inputSplitMaxSizeMB);
 
   static FlinkPipelineOptions defaults() {
     return PipelineOptionsFactory.as(FlinkPipelineOptions.class);

runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java

@@ -113,12 +113,12 @@ public float getAverageRecordWidth() {
   private long getDesiredSizeBytes(int numSplits) throws Exception {
     long totalSize = initialSource.getEstimatedSizeBytes(options);
     long defaultSplitSize = totalSize / numSplits;
-    if (initialSource instanceof FileBasedSource) {
-      long maxSplitSize = options.as(FlinkPipelineOptions.class).getFileInputSplitMaxSizeBytes();
+    long maxSplitSize = options.as(FlinkPipelineOptions.class).getFileInputSplitMaxSizeMB();
+    if (initialSource instanceof FileBasedSource && maxSplitSize > 0) {
       // Most of the time parallelism is < number of files in source.
       // Each file becomes a unique split which commonly create skew.
-      // This limits the size of splits to 128Mb to reduce skew.
-      return Math.min(defaultSplitSize, maxSplitSize);
+      // This limits the size of splits to reduce skew.
+      return Math.min(defaultSplitSize, maxSplitSize * 1024 * 1024);
     } else {
       return defaultSplitSize;
     }