igalshilman commented on a change in pull request #108:
URL: https://github.com/apache/flink-statefun/pull/108#discussion_r425733775
##########
File path:
statefun-e2e-tests/statefun-e2e-tests-common/src/main/java/org/apache/flink/statefun/e2e/common/StatefulFunctionsAppContainers.java
##########
@@ -186,124 +148,220 @@ protected void before() throws Throwable {
protected void after() {
master.stop();
workers.forEach(GenericContainer::stop);
- }
- private static ImageFromDockerfile appImage(
- String appName,
- Configuration dynamicProperties,
- List<ClasspathBuildContextFile> classpathBuildContextFiles) {
- final Path targetDirPath = Paths.get(System.getProperty("user.dir") +
"/target/");
- LOG.info("Building app image with built artifacts located at: {}",
targetDirPath);
-
- final ImageFromDockerfile appImage =
- new ImageFromDockerfile(appName)
- .withFileFromClasspath("Dockerfile", "Dockerfile")
- .withFileFromPath(".", targetDirPath);
-
- Configuration flinkConf = resolveFlinkConf(dynamicProperties);
- String flinkConfString = flinkConfigAsString(flinkConf);
- LOG.info(
- "Resolved Flink configuration after merging dynamic properties with
base flink-conf.yaml:\n\n{}",
- flinkConf);
- appImage.withFileFromString("flink-conf.yaml", flinkConfString);
-
- for (ClasspathBuildContextFile classpathBuildContextFile :
classpathBuildContextFiles) {
- appImage.withFileFromClasspath(
- classpathBuildContextFile.buildContextPath,
classpathBuildContextFile.fromResourcePath);
- }
+ FileUtils.deleteDirectoryQuietly(checkpointDir);
+ }
- return appImage;
+ /** @return the exposed port on master for calling REST APIs. */
+ public int getMasterRestPort() {
+ return master.getMappedPort(8081);
}
/**
- * Merges set dynamic properties with configuration in the base
flink-conf.yaml located in
- * resources.
+ * Restarts a single worker of this Stateful Functions application.
+ *
+ * @param workerIndex the index of the worker to restart.
*/
- private static Configuration resolveFlinkConf(Configuration
dynamicProperties) {
- final InputStream baseFlinkConfResourceInputStream =
-
StatefulFunctionsAppContainers.class.getResourceAsStream("/flink-conf.yaml");
- if (baseFlinkConfResourceInputStream == null) {
- throw new RuntimeException("Base flink-conf.yaml cannot be found.");
+ public void restartWorker(int workerIndex) {
+ if (workerIndex >= workers.size()) {
+ throw new IndexOutOfBoundsException(
+ "Invalid worker index; valid values are 0 to " + (workers.size() -
1));
}
- final File tempBaseFlinkConfFile =
copyToTempFlinkConfFile(baseFlinkConfResourceInputStream);
- return GlobalConfiguration.loadConfiguration(
- tempBaseFlinkConfFile.getParentFile().getAbsolutePath(),
dynamicProperties);
+ final GenericContainer<?> worker = workers.get(workerIndex);
+ worker.stop();
+ worker.start();
+ }
+
+ private static File temporaryCheckpointDir() throws IOException {
Review comment:
The difference is very subtle, where the named volume (on OSX, and
probably on Windows) is created on the same filesystem as the containers (in
the Linux VM that runs docker engine) and hence stuff like sharing a unix
domain socket should work (AFIK)
Also It seems cleaner to let docker handle the whole thing and not touching
the host file system directly.
But I understand the limitation, and wouldn't want to overcomplicated this!
so good to merge!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
