diff --git a/release/src/main/scripts/set_version.sh b/release/src/main/scripts/set_version.sh
index 1789f8e..72af526 100755
--- a/release/src/main/scripts/set_version.sh
+++ b/release/src/main/scripts/set_version.sh
@@ -77,7 +77,8 @@
   sed -i -e "s/sdk_version=.*/sdk_version=$TARGET_VERSION/" gradle.properties
   sed -i -e "s/SdkVersion = .*/SdkVersion = \"$TARGET_VERSION\"/" sdks/go/pkg/beam/core/core.go
   # TODO: [BEAM-4767]
-  sed -i -e "s/'dataflow.container_version' : .*/'dataflow.container_version' : 'beam-${TARGET_VERSION}'/" runners/google-cloud-dataflow-java/build.gradle
+  sed -i -e "s/'dataflow.fnapi_container_version' : .*/'dataflow.fnapi_container_version' : 'beam-${TARGET_VERSION}'/" runners/google-cloud-dataflow-java/build.gradle
+  sed -i -e "s/'dataflow.legacy_container_version' : .*/'dataflow.legacy_container_version' : 'beam-${TARGET_VERSION}'/" runners/google-cloud-dataflow-java/build.gradle
 else
   # For snapshot version:
   #   Java/gradle appends -SNAPSHOT
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/Environments.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/Environments.java
index 222818d..a9232e3 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/Environments.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/Environments.java
@@ -90,20 +90,22 @@
           .build();
 
   public enum JavaVersion {
-    v8("java8", "1.8"),
-    v11("java11", "11");
+    java8("java", "1.8"),
+    java11("java11", "11");
 
-    private final String name;
+    // Legacy name, as used in container image
+    private final String legacyName;
+
+    // Specification version (e.g. System java.specification.version)
     private final String specification;
 
-    JavaVersion(final String name, final String specification) {
-      this.name = name;
+    JavaVersion(final String legacyName, final String specification) {
+      this.legacyName = legacyName;
       this.specification = specification;
     }
 
-    @Override
-    public String toString() {
-      return this.name;
+    public String legacyName() {
+      return this.legacyName;
     }
 
     public String specification() {
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EnvironmentsTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EnvironmentsTest.java
index 082aac6..d980d4a 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EnvironmentsTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/EnvironmentsTest.java
@@ -254,14 +254,14 @@
 
   @Test
   public void testJavaVersion() {
-    assertEquals(JavaVersion.v8, JavaVersion.forSpecification("1.8"));
-    assertEquals("java8", JavaVersion.v8.toString());
-    assertEquals(JavaVersion.v11, JavaVersion.forSpecification("11"));
-    assertEquals("java11", JavaVersion.v11.toString());
+    assertEquals(JavaVersion.java8, JavaVersion.forSpecification("1.8"));
+    assertEquals("java", JavaVersion.java8.legacyName());
+    assertEquals(JavaVersion.java11, JavaVersion.forSpecification("11"));
+    assertEquals("java11", JavaVersion.java11.legacyName());
   }
 
   @Test(expected = UnsupportedOperationException.class)
   public void testJavaVersionInvalid() {
-    assertEquals(JavaVersion.v8, JavaVersion.forSpecification("invalid"));
+    assertEquals(JavaVersion.java8, JavaVersion.forSpecification("invalid"));
   }
 }
diff --git a/runners/google-cloud-dataflow-java/build.gradle b/runners/google-cloud-dataflow-java/build.gradle
index 240318f..6175c30 100644
--- a/runners/google-cloud-dataflow-java/build.gradle
+++ b/runners/google-cloud-dataflow-java/build.gradle
@@ -45,7 +45,8 @@
   filter org.apache.tools.ant.filters.ReplaceTokens, tokens: [
     'dataflow.legacy_environment_major_version' : '8',
     'dataflow.fnapi_environment_major_version' : '8',
-    'dataflow.container_version' : 'beam-2.30.0',
+    'dataflow.legacy_container_version' : 'beam-2.30.0',
+    'dataflow.fnapi_container_version' : 'beam-2.30.0',
     'dataflow.container_base_repository' : 'gcr.io/cloud-dataflow/v1beta3',
   ]
 }
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index 68363a8..fa7c1ec 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -148,6 +148,7 @@
 import org.apache.beam.sdk.util.InstanceBuilder;
 import org.apache.beam.sdk.util.MimeTypes;
 import org.apache.beam.sdk.util.NameUtils;
+import org.apache.beam.sdk.util.ReleaseInfo;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.common.ReflectHelpers;
 import org.apache.beam.sdk.values.KV;
@@ -382,7 +383,7 @@
 
     // Adding the Java version to the SDK name for user's and support convenience.
     String agentJavaVer =
-        (Environments.getJavaVersion() == Environments.JavaVersion.v8)
+        (Environments.getJavaVersion() == Environments.JavaVersion.java8)
             ? "(JRE 8 environment)"
             : "(JDK 11 environment)";
 
@@ -2181,7 +2182,6 @@
       return getDefaultContainerImageUrl(options);
     } else if (containerImage.contains("IMAGE")) {
       // Replace placeholder with default image name
-      // TODO(emilymye): See if we can remove this placeholder
       return containerImage.replace("IMAGE", getDefaultContainerImageNameForJob(options));
     } else {
       return containerImage;
@@ -2195,27 +2195,38 @@
         "%s/%s:%s",
         dataflowRunnerInfo.getContainerImageBaseRepository(),
         getDefaultContainerImageNameForJob(options),
-        dataflowRunnerInfo.getContainerVersion());
+        getDefaultContainerVersion(options));
   }
 
   /**
-   * Construct the default Dataflow container image name based on pipeline type and Environment Java
-   * version.
+   * Construct the default Dataflow container image name based on pipeline type and Java version.
    */
   static String getDefaultContainerImageNameForJob(DataflowPipelineOptions options) {
     Environments.JavaVersion javaVersion = Environments.getJavaVersion();
-    String legacyJavaVersionId =
-        (javaVersion == Environments.JavaVersion.v8) ? "java" : javaVersion.toString();
-
     if (useUnifiedWorker(options)) {
-      return "java";
+      return String.format("beam_%s_sdk", javaVersion.name());
     } else if (options.isStreaming()) {
-      return String.format("beam-%s-streaming", legacyJavaVersionId);
+      return String.format("beam-%s-streaming", javaVersion.legacyName());
     } else {
-      return String.format("beam-%s-batch", legacyJavaVersionId);
+      return String.format("beam-%s-batch", javaVersion.legacyName());
     }
   }
 
+  /**
+   * Construct the default Dataflow container image name based on pipeline type and Java version.
+   */
+  static String getDefaultContainerVersion(DataflowPipelineOptions options) {
+    DataflowRunnerInfo dataflowRunnerInfo = DataflowRunnerInfo.getDataflowRunnerInfo();
+    ReleaseInfo releaseInfo = ReleaseInfo.getReleaseInfo();
+    if (releaseInfo.isDevSdkVersion()) {
+      if (useUnifiedWorker(options)) {
+        return dataflowRunnerInfo.getFnApiDevContainerVersion();
+      }
+      return dataflowRunnerInfo.getLegacyDevContainerVersion();
+    }
+    return releaseInfo.getSdkVersion();
+  }
+
   static boolean useUnifiedWorker(DataflowPipelineOptions options) {
     return hasExperiment(options, "beam_fn_api")
         || hasExperiment(options, "use_runner_v2")
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java
index 4c5b272..fed503d 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java
@@ -41,7 +41,8 @@
       "fnapi.environment.major.version";
   private static final String LEGACY_ENVIRONMENT_MAJOR_VERSION_KEY =
       "legacy.environment.major.version";
-  private static final String CONTAINER_VERSION_KEY = "container.version";
+  private static final String CONTAINER_FNAPI_VERSION_KEY = "fnapi.container.version";
+  private static final String CONTAINER_LEGACY_VERSION_KEY = "legacy.container.version";
   private static final String CONTAINER_BASE_REPOSITORY_KEY = "container.base_repository";
 
   private static class LazyInit {
@@ -100,10 +101,18 @@
     return properties.get(FNAPI_ENVIRONMENT_MAJOR_VERSION_KEY);
   }
 
-  /** Provides the version/tag for constructing the container image path. */
-  public String getContainerVersion() {
-    checkState(properties.containsKey(CONTAINER_VERSION_KEY), "Unknown container version");
-    return properties.get(CONTAINER_VERSION_KEY);
+  /** Provides the version/tag for dev SDK FnAPI container image. */
+  public String getFnApiDevContainerVersion() {
+    checkState(
+        properties.containsKey(CONTAINER_FNAPI_VERSION_KEY), "Unknown FnAPI container version");
+    return properties.get(CONTAINER_FNAPI_VERSION_KEY);
+  }
+
+  /** Provides the version/tag for legacy SDK FnAPI container image. */
+  public String getLegacyDevContainerVersion() {
+    checkState(
+        properties.containsKey(CONTAINER_LEGACY_VERSION_KEY), "Unknown legacy container version");
+    return properties.get(CONTAINER_LEGACY_VERSION_KEY);
   }
 
   /** Provides the version/tag for constructing the container image path. */
diff --git a/runners/google-cloud-dataflow-java/src/main/resources/org/apache/beam/runners/dataflow/dataflow.properties b/runners/google-cloud-dataflow-java/src/main/resources/org/apache/beam/runners/dataflow/dataflow.properties
index 58e92ee..956c447 100644
--- a/runners/google-cloud-dataflow-java/src/main/resources/org/apache/beam/runners/dataflow/dataflow.properties
+++ b/runners/google-cloud-dataflow-java/src/main/resources/org/apache/beam/runners/dataflow/dataflow.properties
@@ -18,5 +18,6 @@
 
 legacy.environment.major.version=@dataflow.legacy_environment_major_version@
 fnapi.environment.major.version=@dataflow.fnapi_environment_major_version@
-container.version=@dataflow.container_version@
+legacy.container.version=@dataflow.legacy_container_version@
+fnapi.container.version=@dataflow.fnapi_container_version@
 container.base_repository=@dataflow.container_base_repository@
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java
index fbb8ed2..e75af29 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java
@@ -49,12 +49,16 @@
         String.format("FnAPI environment major version number %s is not a number", version),
         version.matches("\\d+"));
 
-    // Validate container version does not contain the property name (indicating it was not filled
-    // in).
+    // Validate container versions do not contain the property name.
     assertThat(
-        "container version invalid",
-        info.getContainerVersion(),
-        not(containsString("dataflow.container_version")));
+        "legacy container version invalid",
+        info.getFnApiDevContainerVersion(),
+        not(containsString("dataflow.legacy_container_version")));
+
+    assertThat(
+        "FnAPI container version invalid",
+        info.getLegacyDevContainerVersion(),
+        not(containsString("dataflow.fnapi_container_version")));
 
     // Validate container base repository does not contain the property name
     // (indicating it was not filled in).
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
index f828531..59d105f 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
@@ -1628,29 +1628,38 @@
   @Test
   public void testGetContainerImageForJobFromOptionWithPlaceholder() {
     DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
-    // When image option contains placeholder, container image should
-    // have placeholder replaced with default image for job.
     options.setSdkContainerImage("gcr.io/IMAGE/foo");
 
-    // batch, legacy
-    options.setExperiments(null);
-    options.setStreaming(false);
-    System.setProperty("java.specification.version", "1.8");
-    assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java-batch/foo"));
-    // batch, legacy, jdk11
-    options.setStreaming(false);
-    System.setProperty("java.specification.version", "11");
-    assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java11-batch/foo"));
-    // streaming, legacy
-    System.setProperty("java.specification.version", "1.8");
-    options.setStreaming(true);
-    assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java-streaming/foo"));
-    // streaming, legacy, jdk11
-    System.setProperty("java.specification.version", "11");
-    assertThat(getContainerImageForJob(options), equalTo("gcr.io/beam-java11-streaming/foo"));
-    // streaming, fnapi
-    options.setExperiments(ImmutableList.of("experiment1", "beam_fn_api"));
-    assertThat(getContainerImageForJob(options), equalTo("gcr.io/java/foo"));
+    for (Environments.JavaVersion javaVersion : Environments.JavaVersion.values()) {
+      System.setProperty("java.specification.version", javaVersion.specification());
+      // batch legacy
+      options.setExperiments(null);
+      options.setStreaming(false);
+      assertThat(
+          getContainerImageForJob(options),
+          equalTo(String.format("gcr.io/beam-%s-batch/foo", javaVersion.legacyName())));
+
+      // streaming, legacy
+      options.setExperiments(null);
+      options.setStreaming(true);
+      assertThat(
+          getContainerImageForJob(options),
+          equalTo(String.format("gcr.io/beam-%s-streaming/foo", javaVersion.legacyName())));
+
+      // batch, FnAPI
+      options.setExperiments(ImmutableList.of("beam_fn_api"));
+      options.setStreaming(false);
+      assertThat(
+          getContainerImageForJob(options),
+          equalTo(String.format("gcr.io/beam_%s_sdk/foo", javaVersion.name())));
+
+      // streaming, FnAPI
+      options.setExperiments(ImmutableList.of("beam_fn_api"));
+      options.setStreaming(true);
+      assertThat(
+          getContainerImageForJob(options),
+          equalTo(String.format("gcr.io/beam_%s_sdk/foo", javaVersion.name())));
+    }
   }
 
   @Test
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ReleaseInfo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ReleaseInfo.java
index c284d462..3f21570 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ReleaseInfo.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ReleaseInfo.java
@@ -54,16 +54,21 @@
     return getProperties().get("name");
   }
 
-  /** Provides the BEAM version. ie: 2.18.0-SNAPSHOT */
+  /** Provides the BEAM version e.g. 2.18.0-SNAPSHOT */
   public String getVersion() {
     return getProperties().get("version");
   }
 
-  /** Provides the SDK version. ie: 2.18.0 or 2.18.0.dev */
+  /** Provides the SDK version. e.g. 2.18.0 or 2.18.0.dev */
   public String getSdkVersion() {
     return getProperties().get("sdk_version");
   }
 
+  /** Returns true if SDK version is a dev version (e.g. 2.18.0.dev) */
+  public boolean isDevSdkVersion() {
+    return getProperties().get("sdk_version").contains("dev");
+  }
+
   /** Provides docker image default root (apache). */
   public String getDefaultDockerRepoRoot() {
     return getProperties().get("docker_image_default_repo_root");