diff --git a/src/main/java/com/zdjizhi/flink/voip/CorrelateApp.java b/src/main/java/com/zdjizhi/flink/voip/CorrelateApp.java
index efb53b6..304e04d 100644
--- a/src/main/java/com/zdjizhi/flink/voip/CorrelateApp.java
+++ b/src/main/java/com/zdjizhi/flink/voip/CorrelateApp.java
@@ -103,6 +103,6 @@ public class CorrelateApp {
                 .union(sipDoubleDirOperator.getSideOutput(SIPPairingFunction.SIP_OUTPUT_TAG))
                 .addSink(producer);
 
-        env.execute("SIP-RTP-CORRELATION");
+        env.execute(config.get(JOB_NAME));
     }
 }
diff --git a/src/main/java/com/zdjizhi/flink/voip/conf/FusionConfigs.java b/src/main/java/com/zdjizhi/flink/voip/conf/FusionConfigs.java
index 86f04b5..926c5a0 100644
--- a/src/main/java/com/zdjizhi/flink/voip/conf/FusionConfigs.java
+++ b/src/main/java/com/zdjizhi/flink/voip/conf/FusionConfigs.java
@@ -95,4 +95,13 @@ public class FusionConfigs {
                     .intType()
                     .defaultValue(6)
                     .withDescription("The interval at which RTP state data should be cleared.");
+
+    /**
+     * Configuration option for specifying the name of a job.
+     */
+    public static final ConfigOption<String> JOB_NAME =
+            ConfigOptions.key("job.name")
+                    .stringType()
+                    .defaultValue("correlation_sip_rtp_session")
+                    .withDescription("The name of current job.");
 }
\ No newline at end of file