FlinkDistributionOverlay extends AbstractContainerOverlay { @Override public void configure(ContainerSpecification container) throws IOException { container.getEnvironmentVariables().put(ENV_FLINK_HOME_DIR, TARGET_ROOT.toString()); addPathRecursively(flinkBinPath, TARGET_ROOT, container); addPathRecursively(flinkConfPath, TARGET_ROOT, container); addPathRecursively(flinkLibPath, TARGET_ROOT, container); } FlinkDistributionOverlay(File flinkBinPath, File flinkConfPath, File flinkLibPath); @Override void configure(ContainerSpecification container); static Builder newBuilder();  }
@Test public void testConfigure() throws Exception { File binFolder = tempFolder.newFolder("bin"); File libFolder = tempFolder.newFolder("lib"); File confFolder = tempFolder.newFolder("conf"); Path[] files = createPaths( tempFolder.getRoot(), "bin/config.sh", "bin/taskmanager.sh", "lib/foo.jar", "lib/A/foo.jar", "lib/B/foo.jar", "lib/B/bar.jar"); ContainerSpecification containerSpecification = new ContainerSpecification(); FlinkDistributionOverlay overlay = new FlinkDistributionOverlay( binFolder, confFolder, libFolder ); overlay.configure(containerSpecification); for(Path file : files) { checkArtifact(containerSpecification, new Path(TARGET_ROOT, file.toString())); } }