diff --git a/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorFactory.java b/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorFactory.java
index 1735eec..254bd87 100644
--- a/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorFactory.java
+++ b/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorFactory.java
@@ -54,8 +54,12 @@
                                                       "parameter");
     public static final QName MESSAGE_STORE_Q = new QName(XMLConfigConstants.NULL_NAMESPACE,
                                                           "messageStore");
+    public static final QName TARGET_ENDPOINT_Q =
+            new QName(XMLConfigConstants.NULL_NAMESPACE, "targetEndpoint");
     private static final QName DESCRIPTION_Q
             = new QName(SynapseConstants.SYNAPSE_NAMESPACE, "description");
+    public static final String FORWARDING_PROCESSOR =
+            "org.apache.synapse.message.processors.forward.ScheduledMessageForwardingProcessor";
 
 
     /**
@@ -88,6 +92,18 @@
             handleException("Can't create Message processor without a name ");
         }
 
+        if (FORWARDING_PROCESSOR.equals(clssAtt.getAttributeValue())) {
+            OMAttribute targetEndpointAtt = elem.getAttribute(TARGET_ENDPOINT_Q);
+
+            if (targetEndpointAtt != null) {
+                assert processor != null;
+                processor.setTargetEndpoint(targetEndpointAtt.getAttributeValue());
+            } else {
+                // This validation is commented due to backward compatibility
+                // handleException("Can't create Message processor without a target endpoint ");
+            }
+        }
+
         OMAttribute storeAtt = elem.getAttribute(MESSAGE_STORE_Q);
 
         if (storeAtt != null) {
diff --git a/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorSerializer.java b/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorSerializer.java
index f2a7d48..a2311ef 100644
--- a/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorSerializer.java
+++ b/modules/core/src/main/java/org/apache/synapse/config/xml/MessageProcessorSerializer.java
@@ -49,6 +49,8 @@
 
     private static final Log log = LogFactory.getLog(MessageProcessorSerializer.class);
 
+    public static final String FORWARDING_PROCESSOR =
+            "org.apache.synapse.message.processors.forward.ScheduledMessageForwardingProcessor";
     protected static final OMFactory fac = OMAbstractFactory.getOMFactory();
     protected static final OMNamespace synNS = SynapseConstants.SYNAPSE_OMNAMESPACE;
     protected static final OMNamespace nullNS = fac.createOMNamespace(
@@ -78,6 +80,16 @@
             handleException("Message store Name not specified");
         }
 
+        if (FORWARDING_PROCESSOR.equals(processor.getClass().getName())) {
+            if (processor.getTargetEndpoint() != null) {
+                processorElem.addAttribute(
+                        fac.createOMAttribute("targetEndpoint", nullNS, processor.getTargetEndpoint()));
+            } else {
+                // This validation is removed to support backward compatibility
+                // handleException("Target Endpoint not specified");
+            }
+        }
+
         if (processor.getMessageStoreName() != null) {
             processorElem.addAttribute(fac.createOMAttribute(
                     "messageStore", nullNS, processor.getMessageStoreName()));
diff --git a/modules/core/src/main/java/org/apache/synapse/message/processors/AbstractMessageProcessor.java b/modules/core/src/main/java/org/apache/synapse/message/processors/AbstractMessageProcessor.java
index b319067..d33b8c8 100644
--- a/modules/core/src/main/java/org/apache/synapse/message/processors/AbstractMessageProcessor.java
+++ b/modules/core/src/main/java/org/apache/synapse/message/processors/AbstractMessageProcessor.java
@@ -47,6 +47,10 @@
 
     protected String fileName;
 
+    /** This attribute is only need for forwarding message processor. However, it here because
+     * then we don't need to implement this in sampling processor with nothing */
+    protected String targetEndpoint;
+
     protected SynapseConfiguration configuration;
 
     protected enum State {
@@ -117,4 +121,12 @@
     public String getFileName() {
         return fileName;
     }
+
+    public void setTargetEndpoint(String targetEndpoint) {
+        this.targetEndpoint = targetEndpoint;
+    }
+
+    public String getTargetEndpoint() {
+        return targetEndpoint;
+    }
 }
\ No newline at end of file
diff --git a/modules/core/src/main/java/org/apache/synapse/message/processors/MessageProcessor.java b/modules/core/src/main/java/org/apache/synapse/message/processors/MessageProcessor.java
index b8af8b2..17ef51c 100644
--- a/modules/core/src/main/java/org/apache/synapse/message/processors/MessageProcessor.java
+++ b/modules/core/src/main/java/org/apache/synapse/message/processors/MessageProcessor.java
@@ -85,4 +85,17 @@
      * @return Name of the file where this artifact is defined
      */
     public String getFileName();
+
+    /**
+     * This method set the target-endpoint associated with the Message Processor. Target-endpoint is a required
+     * parameter for Message Forwarding Processor but optional for Sampling Processor.
+     * @param targetEndpoint is the name of the associated endpoint
+     */
+    void setTargetEndpoint(String targetEndpoint);
+
+    /**
+     * This method is used to retrieve the associated target endpoint name of the message processor.
+     * @return The name of the endpoint
+     */
+    String getTargetEndpoint();
 }
diff --git a/modules/core/src/main/java/org/apache/synapse/message/processors/ScheduledMessageProcessor.java b/modules/core/src/main/java/org/apache/synapse/message/processors/ScheduledMessageProcessor.java
index c748ee2..eeb4eed 100644
--- a/modules/core/src/main/java/org/apache/synapse/message/processors/ScheduledMessageProcessor.java
+++ b/modules/core/src/main/java/org/apache/synapse/message/processors/ScheduledMessageProcessor.java
@@ -21,6 +21,7 @@
 
 import org.apache.synapse.SynapseException;
 import org.apache.synapse.core.SynapseEnvironment;
+import org.apache.synapse.message.processors.forward.ForwardingProcessorConstants;
 import org.quartz.*;
 import org.quartz.impl.StdSchedulerFactory;
 
@@ -84,6 +85,7 @@
         JobDataMap jobDataMap = getJobDataMap();
         jobDataMap.put(MessageProcessorConstants.MESSAGE_STORE,
                        configuration.getMessageStore(messageStore));
+        jobDataMap.put(ForwardingProcessorConstants.TARGET_ENDPOINT, getTargetEndpoint());
         jobDataMap.put(MessageProcessorConstants.PARAMETERS, parameters);
 
         JobDetail jobDetail = jobBuilder.usingJobData(jobDataMap).build();
diff --git a/modules/core/src/main/java/org/apache/synapse/message/processors/forward/ForwardingJob.java b/modules/core/src/main/java/org/apache/synapse/message/processors/forward/ForwardingJob.java
index 94ddf4f..aee0d08 100644
--- a/modules/core/src/main/java/org/apache/synapse/message/processors/forward/ForwardingJob.java
+++ b/modules/core/src/main/java/org/apache/synapse/message/processors/forward/ForwardingJob.java
@@ -65,6 +65,7 @@
     private MessageStore messageStore;
     private Axis2BlockingClient sender;
     private ScheduledMessageForwardingProcessor processor;
+    private String targetEndpoint = null;
 
     public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
         //Get the Global Objects from DataMap
@@ -103,6 +104,9 @@
                         .get(ForwardingProcessorConstants.RETRY_HTTP_STATUS_CODES).toString().split(",");
             }
             setSequences(parameters);
+            if (jdm.get(ForwardingProcessorConstants.TARGET_ENDPOINT) != null) {
+                targetEndpoint = (String) jdm.get(ForwardingProcessorConstants.TARGET_ENDPOINT);
+            }
         }
     }
 
@@ -233,15 +237,17 @@
 
     private void handleNewMessage(MessageContext inMsgCtx) {
         sanitizeMsgContext(inMsgCtx);
-        String targetEp = (String) inMsgCtx.getProperty(ForwardingProcessorConstants.TARGET_ENDPOINT);
-        if (targetEp != null) {
-            Endpoint ep = inMsgCtx.getEndpoint(targetEp);
+        if (targetEndpoint == null) {
+            targetEndpoint = (String) inMsgCtx.getProperty(ForwardingProcessorConstants.TARGET_ENDPOINT);
+        }
+        if (targetEndpoint != null) {
+            Endpoint ep = inMsgCtx.getEndpoint(targetEndpoint);
             // stop processing if endpoint is not ready to send
             if(ep.getContext().readyToSend()) {
                 if ((ep != null) && (((AbstractEndpoint) ep).isLeafEndpoint())) {
                     sendMsgToEndpoint(inMsgCtx, ep);
                 } else {
-                    logMsg(targetEp, ep);
+                    logMsg(targetEndpoint, ep);
                     messageStore.poll();
                 }
             }