Dmaap micro service jar

Creating a service running in a standalone jar to handle
the publishing and recieving of dmaap messages for appc.
Dmaap adapter and event listener code is copeid from the
main appc project. It will be moved in a later commit.

Change-Id: I3fa7b5dc60345f0f38f763a243150b8472f985ac
Signed-off-by: Patrick Brady <patrick.brady@att.com>
Issue-ID: APPC-1744

10 files changed, 589 insertions, 0 deletions

diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/.gitignore b/services/appc-dmaap-service/appc-dmaap-event-service/.gitignore
new file mode 100644
index 000000000..b83d22266
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/.gitignore
@@ -0,0 +1 @@
+/target/
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/pom.xml b/services/appc-dmaap-service/appc-dmaap-event-service/pom.xml
new file mode 100644
index 000000000..aa94e45a6
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/pom.xml
@@ -0,0 +1,90 @@
+<!--
+  ============LICENSE_START=======================================================
+  ONAP : APPC
+  ================================================================================
+  Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+  ================================================================================
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+  
+       http://www.apache.org/licenses/LICENSE-2.0
+  
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+  ============LICENSE_END=========================================================
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>org.onap.appc.services.dmaap</groupId>
+  <artifactId>dmaap-event-service</artifactId>
+  <packaging>jar</packaging>
+  <version>1.7.0-SNAPSHOT</version>
+  <name>dmaapService</name>
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.1.6.RELEASE</version>
+    </parent>
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-security</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.2</version>
+        </dependency>
+        <dependency>
+            <groupId>org.onap.appc.services.dmaap</groupId>
+            <artifactId>appc-event-listener-bundle</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.onap.appc.services.dmaap</groupId>
+            <artifactId>appc-message-adapter-factory</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.onap.appc.services.dmaap</groupId>
+            <artifactId>appc-message-adapter-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.onap.appc.services.dmaap</groupId>
+            <artifactId>appc-dmaap-adapter-bundle</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+        </dependency>
+    </dependencies>
+      <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/App.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/App.java
new file mode 100644
index 000000000..afcb3277d
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/App.java
@@ -0,0 +1,57 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * 
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.appc.services.dmaapService;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Properties;
+import java.util.Set;
+import org.onap.appc.listener.AppcEventListenerActivator;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+
+/**
+ * Hello world!
+ *
+ */
+@SpringBootApplication
+public class App 
+{
+    public static void main( String[] args )
+    {
+        
+        System.out.println("Starting");
+        AppcEventListenerActivator event = new AppcEventListenerActivator();
+        try {
+            event.start();
+        } catch (Exception e) {
+            // TODO Auto-generated catch block
+            e.printStackTrace();
+        }
+        
+        SpringApplication.run(App.class, args);
+    }
+    
+
+}
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/AuthenticationConfig.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/AuthenticationConfig.java
new file mode 100644
index 000000000..4031a5144
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/AuthenticationConfig.java
@@ -0,0 +1,46 @@
+package org.onap.appc.services.dmaapService;
+
+import java.util.Properties;
+
+import org.onap.appc.configuration.ConfigurationFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.annotation.authentication.builders.AuthenticationManagerBuilder;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
+import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
+import org.springframework.security.crypto.password.PasswordEncoder;
+
+@Configuration
+public class AuthenticationConfig extends WebSecurityConfigurerAdapter {
+    
+    private final String PROPERTIES_PREFIX = "appc.srvcomm.messaging";
+    private final String DEFAULT_USER = "appc";
+    private final String DEFAULT_PASSWORD = "onapappc";
+    
+    @Override
+    protected void configure(HttpSecurity http) throws Exception {
+        http.csrf().disable().authorizeRequests().anyRequest().authenticated().and().httpBasic();
+    }
+    
+    @Override
+    protected void configure(AuthenticationManagerBuilder auth) throws Exception {
+        org.onap.appc.configuration.Configuration configuration = ConfigurationFactory.getConfiguration();
+        Properties props = configuration.getProperties();
+        String user = props.getProperty(PROPERTIES_PREFIX + ".user");
+        String pass = props.getProperty(PROPERTIES_PREFIX + ".pass");
+        if(user == null) {
+            user = DEFAULT_USER;
+        }
+        if(pass == null) {
+            pass = DEFAULT_PASSWORD;
+        }
+        auth.inMemoryAuthentication().withUser(user).password(encoder().encode(pass)).roles("USER");
+    }
+    
+    @Bean
+    public PasswordEncoder encoder() {
+        return new BCryptPasswordEncoder();
+    }
+
+}
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/EventServer.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/EventServer.java
new file mode 100644
index 000000000..9df67cbf2
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/EventServer.java
@@ -0,0 +1,54 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * 
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.appc.services.dmaapService;
+
+import org.onap.appc.adapter.messaging.dmaap.http.HttpDmaapConsumerImpl;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+@RestController
+public class EventServer {
+    
+    private static final EELFLogger LOG = EELFManager.getInstance().getLogger(EventServer.class);
+
+    @Autowired
+    private PublishService publishService;
+
+    @RequestMapping("/publish")
+    public PublishResponse publish(@RequestBody String body) {
+        PublishRequest req = PublishRequest.parsePublishRequest(body);
+        String result = publishService.publishMessage(req.getProps(),req.getPartition(),
+                req.getTopic(), req.getMessage());
+        if(result.equals("Success")) {
+            return new PublishResponse(result);
+        }
+        LOG.error("Error during message publish: " + result);
+        throw new MessagingException();
+    }
+
+}
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/MessagingException.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/MessagingException.java
new file mode 100644
index 000000000..3c114a1ed
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/MessagingException.java
@@ -0,0 +1,31 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * 
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.appc.services.dmaapService;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.ResponseStatus;
+
+@ResponseStatus(value=HttpStatus.INTERNAL_SERVER_ERROR, reason="Error Publishing Messsage")
+public class MessagingException extends RuntimeException {
+    
+
+}
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishRequest.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishRequest.java
new file mode 100644
index 000000000..c242fdbfd
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishRequest.java
@@ -0,0 +1,83 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * 
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.appc.services.dmaapService;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+public class PublishRequest {
+    
+    private static final EELFLogger LOG = EELFManager.getInstance().getLogger(PublishRequest.class);
+    
+    private String props;
+    private String partition;
+    private String message;
+    private String topic;
+
+    private PublishRequest(String props, String partition, String topic, String message) {
+        this.props = props;
+        this.partition = partition;
+        this.message = message;
+        this.topic = topic;
+    }
+
+    public String getProps() {
+        return props;
+    }
+
+    public String getPartition() {
+        return partition;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public static PublishRequest parsePublishRequest(String data) {
+        //body content: props, partition, topic, message
+        String[] bodyParameters = new String[4];
+        LOG.debug("Parsing message into " + bodyParameters.length + " parts: " + data);
+        int[] bodyParameterSizes = new int[bodyParameters.length];
+        for(int i = 0; i < bodyParameters.length; i ++) {
+            String[] split = data.split("\\.", 2);
+            try {
+                bodyParameterSizes[i] = Integer.parseInt(split[0]);
+            } catch(NumberFormatException e) {
+                LOG.error("Could not parse message: " + data);
+            }
+            data = split[1];
+        }
+        int cursor = 0;
+        for(int i = 0; i < bodyParameters.length; i ++) {
+            if(bodyParameterSizes[i] > 0) {
+                bodyParameters[i] = data.substring(cursor, cursor + bodyParameterSizes[i]);
+                cursor = cursor + bodyParameterSizes[i];
+            }
+        }
+        return new PublishRequest(bodyParameters[0], bodyParameters[1], bodyParameters[2], bodyParameters[3]);
+    }
+
+}
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishResponse.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishResponse.java
new file mode 100644
index 000000000..576c80d80
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishResponse.java
@@ -0,0 +1,36 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * 
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.appc.services.dmaapService;
+
+public class PublishResponse {
+
+    private String status;
+
+    public PublishResponse(String status) {
+        this.status = status;
+    }
+
+    public String getName() {
+        return status;
+    }
+
+}
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishService.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishService.java
new file mode 100644
index 000000000..1e1b9e7f8
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/main/java/org/onap/appc/services/dmaapService/PublishService.java
@@ -0,0 +1,109 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * 
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.appc.services.dmaapService;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Properties;
+
+import org.onap.appc.adapter.factory.DmaapMessageAdapterFactoryImpl;
+import org.onap.appc.configuration.Configuration;
+import org.onap.appc.configuration.ConfigurationFactory;
+import org.springframework.stereotype.Service;
+import org.onap.appc.adapter.message.MessageAdapterFactory;
+import org.onap.appc.adapter.message.Producer;
+import org.onap.appc.adapter.messaging.dmaap.http.HttpDmaapProducerImpl;
+
+@Service
+public class PublishService {
+    
+    private Map<String,Producer> producers;
+    private MessageAdapterFactory factory;
+    Configuration configuration;
+    
+    public PublishService() {
+        this.factory = new DmaapMessageAdapterFactoryImpl();
+        producers = new HashMap<>();
+    }
+    
+    public PublishService(MessageAdapterFactory factory) {
+        this.factory = factory;
+        producers = new HashMap<>();
+    }
+    
+    public String publishMessage(String key, String partition, String topic, String message) {
+        Producer producer = getProducer(key, topic);
+        if(producer == null) {
+            return "Could not find producer with property prefix: " + key;
+        }
+        boolean success = producer.post(partition, message);
+        if(success) {
+            return "Success";
+        }
+        return "Failed. See dmaap service jar log.";
+    }
+    
+    private Producer getProducer(String key, String topic) {
+        String searchKey = key;
+        if(topic != null) {
+            searchKey += topic;
+        }
+        Producer producer = producers.get(searchKey);
+        if(producer != null) {
+            return producer;
+        }
+        producer =  newProducer(key, topic);
+        producers.put(searchKey,producer);
+        return producer;
+    }
+    
+    private Producer newProducer(String key, String topic) {
+        Configuration configuration;
+        if(this.configuration != null) {
+            configuration = this.configuration;
+        } else {
+            configuration = ConfigurationFactory.getConfiguration();
+        }
+        Properties props = configuration.getProperties();
+        HashSet<String> pool = new HashSet<>();
+        if (props != null) {
+            String writeTopic;
+            if(topic == null) {
+                writeTopic = props.getProperty(key + ".topic.write");
+            } else {
+                writeTopic = topic;
+            }
+            String apiKey = props.getProperty(key + ".client.key");
+            String apiSecret = props.getProperty(key + ".client.secret");
+            String hostnames = props.getProperty(key + ".poolMembers");
+            if (hostnames != null && !hostnames.isEmpty()) {
+                for (String name : hostnames.split(",")) {
+                    pool.add(name);
+                }
+            }
+            return factory.createProducer(pool, writeTopic, apiKey, apiSecret);
+        }
+        return null;
+    }
+    
+}
diff --git a/services/appc-dmaap-service/appc-dmaap-event-service/src/test/java/org/onap/appc/services/PublishRequestTest.java b/services/appc-dmaap-service/appc-dmaap-event-service/src/test/java/org/onap/appc/services/PublishRequestTest.java
new file mode 100644
index 000000000..7cc9bccfd
--- /dev/null
+++ b/services/appc-dmaap-service/appc-dmaap-event-service/src/test/java/org/onap/appc/services/PublishRequestTest.java
@@ -0,0 +1,82 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * 
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.appc.services;
+
+import org.onap.appc.services.dmaapService.PublishRequest;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PublishRequestTest {
+    
+    @Test
+    public void testParsePublishRequest() {
+        //props, partition, topic, message
+        String props = "props.test";
+        String partition = "testPartition";
+        String topic = "testTopic";
+        String message = "sameple message data";
+        String data = props.length() + "." + partition.length() + "." + topic.length()
+        + "." + message.length() + "." + props + partition + topic + message;
+        PublishRequest publishRequest = PublishRequest.parsePublishRequest(data);
+        Assert.assertTrue(props.equals(publishRequest.getProps()));
+        Assert.assertTrue(partition.equals(publishRequest.getPartition()));
+        Assert.assertTrue(topic.equals(publishRequest.getTopic()));
+        Assert.assertTrue(message.equals(publishRequest.getMessage()));
+        
+    }
+    
+    @Test
+    public void testParsePublishRequest_period_in_message() {
+        //props, partition, topic, message
+        String props = "props.test";
+        String partition = "testPartition";
+        String topic = "testTopic";
+        String message = "sameple. message. dat.a";
+        String data = props.length() + "." + partition.length() + "." + topic.length()
+        + "." + message.length() + "." + props + partition + topic + message;
+        PublishRequest publishRequest = PublishRequest.parsePublishRequest(data);
+        Assert.assertTrue(props.equals(publishRequest.getProps()));
+        Assert.assertTrue(partition.equals(publishRequest.getPartition()));
+        Assert.assertTrue(topic.equals(publishRequest.getTopic()));
+        Assert.assertTrue(message.equals(publishRequest.getMessage()));
+        
+    }
+    
+    @Test
+    public void testParsePublishRequest_with_missing_value() {
+        //props, partition, topic, message
+        String props = "props.test";
+        String partition = "testPartition";
+        String topic = "";
+        String message = "sameple message data";
+        String data = props.length() + "." + partition.length() + "." + topic.length()
+        + "." + message.length() + "." + props + partition + topic + message;
+        PublishRequest publishRequest = PublishRequest.parsePublishRequest(data);
+        Assert.assertTrue(props.equals(publishRequest.getProps()));
+        Assert.assertTrue(partition.equals(publishRequest.getPartition()));
+        Assert.assertTrue(null == publishRequest.getTopic());
+        Assert.assertTrue(message.equals(publishRequest.getMessage()));
+        
+    }
+
+}