PackageUtil implements Closeable { List<DataflowPackage> stageClasspathElements( Collection<StagedFile> classpathElements, String stagingPath, CreateOptions createOptions) { return stageClasspathElements(classpathElements, stagingPath, DEFAULT_SLEEPER, createOptions); } private  PackageUtil(ExecutorService executorService); static PackageUtil withDefaultThreadPool(); static PackageUtil withExecutorService(ExecutorService executorService); @Override void close(); CompletionStage<StagingResult> stagePackage(
      final PackageAttributes attributes,
      final Sleeper retrySleeper,
      final CreateOptions createOptions); DataflowPackage stageToFile(
      byte[] bytes, String target, String stagingPath, CreateOptions createOptions);  }
@Test public void testPackageUploadWithFileSucceeds() throws Exception { Pipe pipe = Pipe.open(); String contents = "This is a test!"; File tmpFile = makeFileWithContents("file.txt", contents); when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create(new FileNotFoundException("some/path")))); when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink()); List<DataflowPackage> targets = defaultPackageUtil.stageClasspathElements( ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())), STAGING_PATH, createOptions); DataflowPackage target = Iterables.getOnlyElement(targets); verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class)); verify(mockGcsUtil).create(any(GcsPath.class), anyString()); verifyNoMoreInteractions(mockGcsUtil); assertThat(target.getName(), endsWith(".txt")); assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName())); assertThat( new LineReader(Channels.newReader(pipe.source(), StandardCharsets.UTF_8.name())).readLine(), equalTo(contents)); }
@Test public void testStagingPreservesClasspath() throws Exception { File smallFile = makeFileWithContents("small.txt", "small"); File largeFile = makeFileWithContents("large.log", "large contents"); when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create(new FileNotFoundException("some/path")))); when(mockGcsUtil.create(any(GcsPath.class), anyString())) .thenAnswer(invocation -> Pipe.open().sink()); List<DataflowPackage> targets = defaultPackageUtil.stageClasspathElements( ImmutableList.of( makeStagedFile(smallFile.getAbsolutePath()), makeStagedFile(largeFile.getAbsolutePath())), STAGING_PATH, createOptions); assertThat(targets.get(0).getName(), endsWith(".txt")); assertThat(targets.get(1).getName(), endsWith(".log")); }
@Test public void testPackageUploadWithDirectorySucceeds() throws Exception { Pipe pipe = Pipe.open(); File tmpDirectory = tmpFolder.newFolder("folder"); tmpFolder.newFolder("folder", "empty_directory"); tmpFolder.newFolder("folder", "directory"); makeFileWithContents("folder/file.txt", "This is a test!"); makeFileWithContents("folder/directory/file.txt", "This is also a test!"); when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create(new FileNotFoundException("some/path")))); when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink()); defaultPackageUtil.stageClasspathElements( ImmutableList.of(makeStagedFile(tmpDirectory.getAbsolutePath())), STAGING_PATH, createOptions); verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class)); verify(mockGcsUtil).create(any(GcsPath.class), anyString()); verifyNoMoreInteractions(mockGcsUtil); List<String> zipEntryNames = new ArrayList<>(); try (ZipInputStream inputStream = new ZipInputStream(Channels.newInputStream(pipe.source()))) { for (ZipEntry entry = inputStream.getNextEntry(); entry != null; entry = inputStream.getNextEntry()) { zipEntryNames.add(entry.getName()); } } assertThat( zipEntryNames, containsInAnyOrder("directory/file.txt", "empty_directory/", "file.txt")); }
@Test public void testPackageUploadWithEmptyDirectorySucceeds() throws Exception { Pipe pipe = Pipe.open(); File tmpDirectory = tmpFolder.newFolder("folder"); when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create(new FileNotFoundException("some/path")))); when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink()); List<DataflowPackage> targets = defaultPackageUtil.stageClasspathElements( ImmutableList.of(makeStagedFile(tmpDirectory.getAbsolutePath())), STAGING_PATH, createOptions); DataflowPackage target = Iterables.getOnlyElement(targets); verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class)); verify(mockGcsUtil).create(any(GcsPath.class), anyString()); verifyNoMoreInteractions(mockGcsUtil); assertThat(target.getName(), endsWith(".jar")); assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName())); try (ZipInputStream zipInputStream = new ZipInputStream(Channels.newInputStream(pipe.source()))) { assertNull(zipInputStream.getNextEntry()); } }
@Test public void testPackageUploadIsSkippedWhenFileAlreadyExists() throws Exception { File tmpFile = makeFileWithContents("file.txt", "This is a test!"); when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create( createStorageObject(STAGING_PATH, tmpFile.length())))); defaultPackageUtil.stageClasspathElements( ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())), STAGING_PATH, createOptions); verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class)); verifyNoMoreInteractions(mockGcsUtil); }
@Test public void testPackageUploadIsNotSkippedWhenSizesAreDifferent() throws Exception { Pipe pipe = Pipe.open(); File tmpDirectory = tmpFolder.newFolder("folder"); tmpFolder.newFolder("folder", "empty_directory"); tmpFolder.newFolder("folder", "directory"); makeFileWithContents("folder/file.txt", "This is a test!"); makeFileWithContents("folder/directory/file.txt", "This is also a test!"); when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create( createStorageObject(STAGING_PATH, Long.MAX_VALUE)))); when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink()); defaultPackageUtil.stageClasspathElements( ImmutableList.of(makeStagedFile(tmpDirectory.getAbsolutePath())), STAGING_PATH, createOptions); verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class)); verify(mockGcsUtil).create(any(GcsPath.class), anyString()); verifyNoMoreInteractions(mockGcsUtil); }
@Test public void testPackageUploadWithExplicitPackageName() throws Exception { Pipe pipe = Pipe.open(); File tmpFile = makeFileWithContents("file.txt", "This is a test!"); final String overriddenName = "alias.txt"; when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create(new FileNotFoundException("some/path")))); when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink()); List<DataflowPackage> targets = defaultPackageUtil.stageClasspathElements( ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath(), overriddenName)), STAGING_PATH, createOptions); DataflowPackage target = Iterables.getOnlyElement(targets); verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class)); verify(mockGcsUtil).create(any(GcsPath.class), anyString()); verifyNoMoreInteractions(mockGcsUtil); assertThat(target.getName(), equalTo(overriddenName)); assertThat(target.getLocation(), RegexMatcher.matches(STAGING_PATH + "alias.txt")); }
@Test public void testPackageUploadIsSkippedWithNonExistentResource() throws Exception { String nonExistentFile = FileSystems.matchNewResource(tmpFolder.getRoot().getPath(), true) .resolve("non-existent-file", StandardResolveOptions.RESOLVE_FILE) .toString(); assertEquals( Collections.EMPTY_LIST, defaultPackageUtil.stageClasspathElements( ImmutableList.of(makeStagedFile(nonExistentFile)), STAGING_PATH, createOptions)); }
@Test public void testPackageUploadWithLargeClasspathLogsWarning() throws Exception { File tmpFile = makeFileWithContents("file.txt", "This is a test!"); when(mockGcsUtil.getObjects(anyListOf(GcsPath.class))) .thenReturn( ImmutableList.of( StorageObjectOrIOException.create( createStorageObject(STAGING_PATH, tmpFile.length())))); List<StagedFile> classpathElements = Lists.newLinkedList(); for (int i = 0; i < 1005; ++i) { String eltName = "element" + i; classpathElements.add(makeStagedFile(tmpFile.getAbsolutePath(), eltName)); } defaultPackageUtil.stageClasspathElements(classpathElements, STAGING_PATH, createOptions); logged.verifyWarn("Your classpath contains 1005 elements, which Google Cloud Dataflow"); }