[GitHub] [flink] 1996fanrui commented on a diff in pull request #23191: [FLINK-32809][yarn] Fixes YarnClusterDescriptor#isArchiveOnlyIncluded…

["1996fanrui (via GitHub)" <gi...@apache.org>]

1996fanrui commented on code in PR #23191:
URL: https://github.com/apache/flink/pull/23191#discussion_r1289738906


##########
flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java:
##########
@@ -305,23 +305,25 @@ public void addShipFiles(List<File> shipFiles) {
     private void addShipArchives(List<File> shipArchives) {
         checkArgument(
                 isArchiveOnlyIncludedInShipArchiveFiles(shipArchives),
-                "Non-archive files are included.");
+                "Not all files included in shipArchives are archive files.");

Review Comment:
   How about this change?
   
   ```suggestion
                   "Directories or non-archive files are included.");
   ```



##########
flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java:
##########
@@ -672,6 +672,30 @@ void testShipUsrLib() throws IOException {
         }
     }
 
+    /** Tests that the {@code YarnConfigOptions.SHIP_ARCHIVES} only supports archive files. */
+    @Test
+    void testShipArchives() throws IOException {
+        final File homeFolder =
+                Files.createTempDirectory(temporaryFolder, UUID.randomUUID().toString()).toFile();
+        File dir1 = new File(homeFolder.getPath(), "dir1");
+        File archive1 = new File(homeFolder.getPath(), "archive1.zip");
+        File archive2 = new File(homeFolder.getPath(), "archive2.zip");
+        assertThat(dir1.createNewFile()).isTrue();

Review Comment:
   Actually, the dir1 is a file instead of directory because you `createNewFile()` here.
   
   We can create a directory by `Files.createTempDirectory()`.



##########
flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java:
##########
@@ -672,6 +672,30 @@ void testShipUsrLib() throws IOException {
         }
     }
 
+    /** Tests that the {@code YarnConfigOptions.SHIP_ARCHIVES} only supports archive files. */
+    @Test
+    void testShipArchives() throws IOException {
+        final File homeFolder =
+                Files.createTempDirectory(temporaryFolder, UUID.randomUUID().toString()).toFile();
+        File dir1 = new File(homeFolder.getPath(), "dir1");
+        File archive1 = new File(homeFolder.getPath(), "archive1.zip");
+        File archive2 = new File(homeFolder.getPath(), "archive2.zip");
+        assertThat(dir1.createNewFile()).isTrue();
+        assertThat(archive1.createNewFile()).isTrue();
+        assertThat(archive2.createNewFile()).isTrue();
+        Configuration flinkConfiguration = new Configuration();
+        flinkConfiguration.set(YarnConfigOptions.SHIP_ARCHIVES,
+                Arrays.asList(dir1.getAbsolutePath(), archive1.getAbsolutePath()));
+
+        assertThrows("Not all files included in shipArchives are archive files.",
+                IllegalArgumentException.class,
+                () -> createYarnClusterDescriptor(flinkConfiguration));
+
+        flinkConfiguration.set(YarnConfigOptions.SHIP_ARCHIVES,
+                Arrays.asList(archive1.getAbsolutePath(), archive2.getAbsolutePath()));
+        createYarnClusterDescriptor(flinkConfiguration);

Review Comment:
   You tested 2 cases:
   1. dir + archiveFile
   2. archiveFile + archiveFile
   
   It's better to add a case: non-archive file + archiveFile.



##########
flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java:
##########
@@ -305,23 +305,25 @@ public void addShipFiles(List<File> shipFiles) {
     private void addShipArchives(List<File> shipArchives) {
         checkArgument(
                 isArchiveOnlyIncludedInShipArchiveFiles(shipArchives),
-                "Non-archive files are included.");
+                "Not all files included in shipArchives are archive files.");
         this.shipArchives.addAll(shipArchives);
     }
 
     private static boolean isArchiveOnlyIncludedInShipArchiveFiles(List<File> shipFiles) {
-        return shipFiles.stream()
+        List<String> shipFileNames = shipFiles.stream()
                 .filter(File::isFile)
                 .map(File::getName)
                 .map(String::toLowerCase)
-                .allMatch(
+                .filter(
                         name ->
                                 name.endsWith(".tar.gz")
                                         || name.endsWith(".tar")
                                         || name.endsWith(".tgz")
                                         || name.endsWith(".dst")
                                         || name.endsWith(".jar")
-                                        || name.endsWith(".zip"));
+                                        || name.endsWith(".zip"))
+                .collect(Collectors.toList());

Review Comment:
   The collect isn't necessary, we can use the `.count()` instead.
   
   And the fieldName can be `archivedFileCount`, and return `archivedFileCount == shipFiles.size()`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org