A little refactoring

Author: shnapz

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileSystems.java

@@ -410,7 +410,8 @@ public ResourceId apply(@Nonnull Metadata input) {
    * <p>- Otherwise, report top level only.
    */
   public static void reportSourceLineage(List<ResourceId> resourceIds) {
-    if (resourceIds.size() <= 100) {
+    final int MAX_LINEAGE_TARGETS = 100;
+    if (resourceIds.size() <= MAX_LINEAGE_TARGETS) {
       for (ResourceId resourceId : resourceIds) {
         FileSystems.reportSourceLineage(resourceId);
       }
@@ -419,7 +420,7 @@ public static void reportSourceLineage(List<ResourceId> resourceIds) {
       for (ResourceId resourceId : resourceIds) {
         ResourceId dir = resourceId.getCurrentDirectory();
         uniqueDirs.add(dir);
-        if (uniqueDirs.size() > 100) {
+        if (uniqueDirs.size() > MAX_LINEAGE_TARGETS) {
           FileSystems.reportSourceLineage(dir, LineageLevel.TOP_LEVEL);
           return;
         }

sdks/java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java

@@ -465,7 +465,6 @@ public Read<K, V> withConfiguration(Configuration configuration) {
       if (getValueTranslationFunction() == null) {
         builder.setValueTypeDescriptor((TypeDescriptor<V>) inputFormatValueClass);
       }
-
       return builder.build();
     }
 
@@ -728,11 +727,8 @@ public List<BoundedSource<KV<K, V>>> split(long desiredBundleSizeBytes, Pipeline
         LOG.info("Not splitting source {} because source is already split.", this);
         return ImmutableList.of(this);
       }
-
       computeSplitsIfNecessary();
-
       reportSourceLineage(inputSplits);
-
       LOG.info(
           "Generated {} splits. Size of first split is {} ",
           inputSplits.size(),
@@ -754,17 +750,13 @@ public List<BoundedSource<KV<K, V>>> split(long desiredBundleSizeBytes, Pipeline
 
     /** Report only file-based sources */
     private void reportSourceLineage(final List<SerializableSplit> inputSplits) {
-      List<ResourceId> fileResources = new ArrayList<>();
-
-      for (SerializableSplit split : inputSplits) {
-        InputSplit inputSplit = split.getSplit();
-
-        if (inputSplit instanceof FileSplit) {
-          String pathString = ((FileSplit) inputSplit).getPath().toString();
-          ResourceId resourceId = FileSystems.matchNewResource(pathString, false);
-          fileResources.add(resourceId);
-        }
-      }
+      List<ResourceId> fileResources =
+          inputSplits.stream()
+              .map(SerializableSplit::getSplit)
+              .filter(FileSplit.class::isInstance)
+              .map(FileSplit.class::cast)
+              .map(fileSplit -> FileSystems.matchNewResource(fileSplit.getPath().toString(), false))
+              .collect(Collectors.toList());
 
       FileSystems.reportSourceLineage(fileResources);
     }