1
\$\begingroup\$

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.java

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();
                sb.append(value).append(seperator);
            });
        } catch (Exception e) {
            throw e;
        }

        return sb.toString();
    }

}

PropertyFileReader.java

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");
            props.load(file);
        } catch (IOException e) {
            System.out.println(e.getMessage());
        }
    }

    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);
    }
}

ElasticWriter.java

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"),
                        Integer.parseInt(PropertyFileReader.getInstance().getProperty("elasticsearchPortNumber")),
                        PropertyFileReader.getInstance().getProperty("elasticsearchScheme")))
                .build();

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

}

App.java

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");
        @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        consumer.subscribe(Arrays.asList(topicName));

        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
                            .asList(PropertyFileReader.getInstance().getProperty(recordType).split(","));
                    String path = PropertyFileReader.getInstance().getProperty(recordType + "_path");
                    String docId = IDGenerator.generate(jsonInput, fields, "_");

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

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }
    }
}

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.

\$\endgroup\$

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.