I consume JSON data from Kafka Topics, and parse the JSON to create a Document ID (Unique Identifier) to persist the JSON into Elasticsearch.


IDGenerator generates a document ID to persist into Elasticsearch.

public class IDGenerator {

    public static String generate(String inputJson, List<String> idParams,final String seperator) throws Exception {
        JsonNode node;
        StringBuilder sb = new StringBuilder();
        try {
            node = new ObjectMapper().readTree(new StringReader(inputJson));
            idParams.forEach(e -> {
                String value = node.findValue(e).asText();
        } catch (Exception e) {
            throw e;

        return sb.toString();



PropertyFileReader reads the properties from the property file and fetches all the data.

public class PropertyFileReader {

    private final Properties props = new Properties();

    private PropertyFileReader() {
        String filePath = new File("").getAbsolutePath();

        FileInputStream file;

        try {
            file = new FileInputStream(filePath + "/config/config.properties");
        } catch (IOException e) {

    private static class PropHolder {
        private static final PropertyFileReader INSTANCE = new PropertyFileReader();

    public static PropertyFileReader getInstance() {
        return PropHolder.INSTANCE;

    public String getProperty(String key) {
        return props.getProperty(key);

    public Set<String> getAllPropertyNames() {
        return props.stringPropertyNames();

    public boolean containsKey(String key) {
        return props.containsKey(key);


This class writes data to Elasticsearch.

public class ElasticWriter {

    public static void write(String path, String docId, List<String> fields, String inputJson) throws Exception {
        RestClient restClient = RestClient
                .builder(new HttpHost(PropertyFileReader.getInstance().getProperty("elasticsearchHostName"),

        Map<String, String> params = Collections.emptyMap();
        HttpEntity entity = new NStringEntity(inputJson, ContentType.APPLICATION_JSON);
        Response response = restClient.performRequest("POST", path + docId, params, entity);



This is the class from where the code gets started.

public class App {
    public static void main(String[] args) {

        String topicName = "csc_alert";
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "StringConsumer");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);


        System.out.println("Subscribed to topic " + topicName);

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
                String jsonInput = record.value();
                try {
                    JsonNode actualObject = new ObjectMapper().readTree(new StringReader(jsonInput));
                    String recordType = actualObject.get("recordType").asText();

                    List<String> fields = Arrays
                    String path = PropertyFileReader.getInstance().getProperty(recordType + "_path");
                    String docId = IDGenerator.generate(jsonInput, fields, "_");

                    ElasticWriter.write(path, docId, fields, jsonInput);

                } catch (Exception e) {


I had a discussion with one of my team members and he says that I need to use Object Oriented Principle rather than using a static way of writing code; also he specified about optimizing the JSON processing.

I want to optimize the code, as it is reading and parsing a single JSON input multiple times. This is a performance block.


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.