diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunner.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunner.java
index e513866c6a79..d4fe5151ba69 100644
--- a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunner.java
+++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunner.java
@@ -232,7 +232,8 @@ public final ProgramController run(final Program program, ProgramOptions oldOpti
 
       // Localize the program jar
       Location programJarLocation = program.getJarLocation();
-      localizeResources.put(programJarLocation.getName(),
+      String programJarName = getProgramJarName(program, oldOptions);
+      localizeResources.put(programJarName,
           new LocalizeResource(programJarLocation.toURI(), false));
 
       // Update the ProgramOptions to carry program and runtime information necessary to reconstruct the program
@@ -342,6 +343,19 @@ public final ProgramController run(final Program program, ProgramOptions oldOpti
     }
   }
 
+  @VisibleForTesting
+  String getProgramJarName(Program program, ProgramOptions options) {
+    String programJarName = program.getJarLocation().getName();
+    if (options.getArguments().hasOption(ProgramOptionConstants.PROGRAM_JAR_HASH)) {
+      // if hash value for program.jar has been provided, we append it to filename.
+      String programJarHash = options.getArguments()
+          .getOption(ProgramOptionConstants.PROGRAM_JAR_HASH);
+      programJarName = programJarName.replace(".jar",
+          String.format("_%s%s", programJarHash, ".jar"));
+    }
+    return programJarName;
+  }
+
   @VisibleForTesting
   void setSchedulerQueue(TwillPreparer twillPreparer, Program program, ProgramOptions options) {
     String schedulerQueueName = options.getArguments()
@@ -356,14 +370,8 @@ void setSchedulerQueue(TwillPreparer twillPreparer, Program program, ProgramOpti
   Map<String, String> getExtraSystemArgs(ProgramLaunchConfig launchConfig,
       Program program, ProgramOptions options) {
     Map<String, String> extraSystemArgs = new HashMap<>(launchConfig.getExtraSystemArguments());
-    String programJarName = program.getJarLocation().getName();
+    String programJarName = getProgramJarName(program, options);
     if (options.getArguments().hasOption(ProgramOptionConstants.PROGRAM_JAR_HASH)) {
-      // if hash value for program.jar has been provided, we append it to filename.
-      String programJarHash = options.getArguments()
-          .getOption(ProgramOptionConstants.PROGRAM_JAR_HASH);
-      programJarName = programJarName.replace(".jar",
-          String.format("_%s%s", programJarHash, ".jar"));
-
       Set<String> cacheableFiles = new HashSet<>();
       if (options.getArguments().hasOption(ProgramOptionConstants.CACHEABLE_FILES)) {
         cacheableFiles =
diff --git a/cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunnerTest.java b/cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunnerTest.java
index abe82cdb2116..b6b4146b78b2 100644
--- a/cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunnerTest.java
+++ b/cdap-app-fabric/src/test/java/io/cdap/cdap/internal/app/runtime/distributed/DistributedProgramRunnerTest.java
@@ -187,6 +187,22 @@ public void testSetSchedulerQueue() {
     Assert.assertEquals("que", preparer.getSchedulerQueue());
   }
 
+  @Test
+  public void testGetProgramJarNameWithoutJarHash() {
+    ProgramOptions options =
+        new SimpleProgramOptions(program.getId(), new BasicArguments(), new BasicArguments());
+    Assert.assertEquals("program.jar", runner.getProgramJarName(program, options));
+  }
+
+  @Test
+  public void testGetProgramJarNameWithJarHash() {
+    BasicArguments systemArgs =
+        new BasicArguments(ImmutableMap.of(ProgramOptionConstants.PROGRAM_JAR_HASH, "1234abc"));
+    ProgramOptions options =
+        new SimpleProgramOptions(program.getId(), systemArgs, new BasicArguments());
+    Assert.assertEquals("program_1234abc.jar", runner.getProgramJarName(program, options));
+  }
+
   private Program createProgram(File baseDir) {
     Application app = new AppWithSparkProgram();
     DefaultAppConfigurer configurer =