1.

		<dependency>
			<groupId>com.googlecode.json-simple</groupId>
			<artifactId>json-simple</artifactId>
			<version>1.1.1</version>
		</dependency>

Object message = kafkaMessage;
JSONParser parser = new JSONParser();
JSONObject messageJsonObject = (JSONObject) parser.parse(message.toString());
String metricsetName = getMemoryRate(messageJsonObject);
String hostname = getHost(messageJsonObject);
Long systemTimestamp = getSystemTimestamp(messageJsonObject);

public static Integer getMemoryRate(JSONObject messageJsonObject) {
    JSONObject system = (JSONObject) messageJsonObject.get("system");
    JSONObject memory = (JSONObject) system.get(METRIC_MEMORY);
    JSONObject actual = (JSONObject) memory.get("actual");
    JSONObject used = (JSONObject) actual.get("used");
    if (!used.get("pct").toString().equals("0")) {
        double pct = (double) used.get("pct") * 100;
        Integer rate = (int) pct;
        return rate;
    } else {
        return 0;
    }
}