Repository: wso2/micro-integrator
Issue: Message Processor state is not saved properly in MI registry

**Description:**
The Activate parameter we configure in synapse configurations is only helpful for the message processor to deploy for the very first time. Once it is deployed, the message processor's state is preserved in the registry [1]. The current state is assigned to the property MESSAGE_PROCESSOR_STATE and saved in the registry (<MI-HOME>/registry/governance/repository/components/org.apache.synapse.message.processor). 

In MI the state of the message processor is not properly stored as a property. It should be stored as key-value pair as shown below,
`MESSAGE_PROCESSOR_STATE=PAUSED`

but the value is stored without the property name.

**Steps to reproduce:**
1. Configure JMS transport for Active MQ by following the documentation [2]
2. Configure JMS Message Store, Scheduled Message Forwarding processor and the Proxy service by following the doc [3]
3. Invoke the Proxy Service

[1] https://github.com/wso2/wso2-synapse/blob/2a1064286a31d2efa458b31da4d518e1f60b6df1/modules/core/src/main/java/org/apache/synapse/message/processor/impl/ScheduledMessageProcessor.java#L564
[2] https://ei.docs.wso2.com/en/latest/micro-integrator/setup/brokers/configure-with-ActiveMQ/
[3] https://ei.docs.wso2.com/en/latest/micro-integrator/use-cases/examples/jms_examples/guaranteed-delivery-with-failover/


Changed files:
components/mediation/registry/org.wso2.micro.integrator.registry/src/main/java/org/wso2/micro/integrator/registry/MicroIntegratorRegistry.java
integration/mediation-tests/tests-platform/tests-coordination/src/test/java/org/wso2/micro/integrator/message/processor/MessageProcessorTests.java

diff --git a/components/mediation/registry/org.wso2.micro.integrator.registry/src/main/java/org/wso2/micro/integrator/registry/MicroIntegratorRegistry.java b/components/mediation/registry/org.wso2.micro.integrator.registry/src/main/java/org/wso2/micro/integrator/registry/MicroIntegratorRegistry.java
index 8f879f8d2..62e0ea968 100644
--- a/components/mediation/registry/org.wso2.micro.integrator.registry/src/main/java/org/wso2/micro/integrator/registry/MicroIntegratorRegistry.java
+++ b/components/mediation/registry/org.wso2.micro.integrator.registry/src/main/java/org/wso2/micro/integrator/registry/MicroIntegratorRegistry.java
@@ -23,6 +23,7 @@ import org.apache.axiom.om.OMException;
 import org.apache.axiom.om.OMNode;
 import org.apache.axiom.om.impl.builder.StAXOMBuilder;
 import org.apache.axiom.om.impl.llom.OMDocumentImpl;
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.synapse.SynapseException;
@@ -55,6 +56,7 @@ import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamReader;
 
 import static org.wso2.micro.integrator.registry.MicroIntegratorRegistryConstants.DEFAULT_MEDIA_TYPE;
+import static org.wso2.micro.integrator.registry.MicroIntegratorRegistryConstants.PROPERTY_EXTENTION;
 import static org.wso2.micro.integrator.registry.MicroIntegratorRegistryConstants.URL_SEPARATOR;
 
 public class MicroIntegratorRegistry extends AbstractRegistry {
@@ -510,21 +512,31 @@ public class MicroIntegratorRegistry extends AbstractRegistry {
             if (isDirectory && !targetPath.endsWith(URL_SEPARATOR)) {
                 targetPath += URL_SEPARATOR;
             }
-
             String parent = getParentPath(targetPath, isDirectory);
-            String fileName = getResourceName(targetPath);
-
-            Properties metadata = new Properties();
-            if (mediaType != null) {
-                metadata.setProperty(METADATA_KEY_MEDIA_TYPE, mediaType);
-            }
-
             try {
-                writeToFile(new URI(parent), fileName, content, metadata);
+                if (isDirectory) {
+                    if (!new File(parent).exists() && !new File(parent).mkdirs()) {
+                        handleException("Unable to create directory: " + parent);
+                    }
+                    if (StringUtils.isNotEmpty(propertyName)) {
+                        writeToFile(new URI(parent), PROPERTY_EXTENTION, propertyName + "=" + content, null);
+                    }
+                } else {
+                    String fileName = getResourceName(targetPath);
+                    if (StringUtils.isEmpty(propertyName)) {
+                        Properties metadata = new Properties();
+                        if (mediaType != null) {
+                            metadata.setProperty(METADATA_KEY_MEDIA_TYPE, mediaType);
+                        }
+                        writeToFile(new URI(parent), fileName, content, metadata);
+                    } else {
+                        writeToFile(new URI(parent), fileName, "", null);
+                        writeToFile(new URI(parent), fileName + PROPERTY_EXTENTION, propertyName + "=" + content, null);
+                    }
+                }
             } catch (Exception e) {
                 handleException("Error when adding a new resource", e);
             }
-
         } else {
             log.warn("Creating new resource in remote registry is NOT SUPPORTED. Unable to create: " + path);
         }
@@ -765,8 +777,9 @@ public class MicroIntegratorRegistry extends AbstractRegistry {
         try (BufferedWriter writer = new BufferedWriter(new FileWriter(newFile))) {
             writer.write(content);
             writer.flush();
-            writeMetadata(parent, newFileName, metadata);
-
+            if (metadata != null) {
+                writeMetadata(parent, newFileName, metadata);
+            }
             if (log.isDebugEnabled()) {
                 log.debug("Successfully content written to file : " + parentName + URL_SEPARATOR + newFileName);
             }
diff --git a/integration/mediation-tests/tests-platform/tests-coordination/src/test/java/org/wso2/micro/integrator/message/processor/MessageProcessorTests.java b/integration/mediation-tests/tests-platform/tests-coordination/src/test/java/org/wso2/micro/integrator/message/processor/MessageProcessorTests.java
index a5c74d38a..1e08bc358 100644
--- a/integration/mediation-tests/tests-platform/tests-coordination/src/test/java/org/wso2/micro/integrator/message/processor/MessageProcessorTests.java
+++ b/integration/mediation-tests/tests-platform/tests-coordination/src/test/java/org/wso2/micro/integrator/message/processor/MessageProcessorTests.java
@@ -231,9 +231,7 @@ public class MessageProcessorTests extends ESBIntegrationTest {
         }
     }
 
-    @Test(dependsOnMethods = { "testMpActivationViaPassiveNode" },
-            // due to https://github.com/wso2/micro-integrator/issues/1071
-            enabled = false)
+    @Test(dependsOnMethods = { "testMpActivationViaPassiveNode" })
     void testMpStateUponRedeployment() throws Exception {
 
         logManager.clearAll();