Static Factories are Great!

Every now and then I jump on classes with multiple constructors or classes that are rigorous to work with. Let alone not being able to mock part of their components and at the end being forced to use reflection for testing (mockito based, old school, you choose).

Imagine a Producer class that you use for Kafka. A class that provides you with some abstraction on sending messages.

package com.gkatzioura.kafka.producer;

import java.util.Properties;

import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j

public class StrMessageProducer {

    private Producer<String,String> producer;

    private String topic = "test-topic";

    StrMessageProducer() {

        var kafkaProperties = new Properties();

        kafkaProperties.put("bootstrap.servers",System.getProperty("bootstrap.servers"));

        kafkaProperties.put("key.serializer",System.getProperty("key.serializer"));

        kafkaProperties.put("value.serialize",System.getProperty("value.serializer"));

        var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);

        this.producer = kafkaProducer;

    }

    public void send(String message) {

        var producerRecord = new ProducerRecord<String,String>(topic,null, message);

        try {

            var metadata = producer.send(producerRecord).get();

            log.info("Submitted {}",metadata.offset());

        }

        catch (InterruptedException |ExecutionException e) {

            log.error("Could not send record",e);

        }

    }

}

Apart from being an ugly class it is also very hard to change some of its components.

For example

◉ I cannot use this class to post to another topic

◉ I cannot use this class to use a different server configuration apart from the one on the properties

◉ It’s difficult to test the functionality of the class since crucial components are created through the constructor

It’s obvious that the constructor in this case serves the purpose on creating a Kafka producer based on the system properties. But the responsibility of the class is to use that producer in order send messages in a specific way. Thus I will move the creation of the Producer from the constructor. Also because we might want to swap the topic used, I will also inject the topic instead of having it hardcoded.

By doing so we encourage dependency injection. We make it easy to swap the ingredients of the class however the execution would be the same.

package com.gkatzioura.kafka.producer;

import java.util.Properties;

import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j

public class StrMessageProducer {

    private final Producer<String,String> producer;

    private final String topic;

    StrMessageProducer(Producer<String,String> producer, String topic) {

        this.producer = producer;

        this.topic = topic;

    }

    public void send(String message) {

        var producerRecord = new ProducerRecord<String,String>(topic,null, message);

        try {

            var metadata = producer.send(producerRecord).get();

            log.info("Submitted {}",metadata.offset());

        }

        catch (InterruptedException |ExecutionException e) {

            log.error("Could not send record",e);

        }

    }

}

But we still need the producer to be created somehow. This is where the factory pattern kicks in.

We shall add static factories in order to have instances of the StrMessageProducer class with different configurations.

Let’s add two factory methods

The first factory method would be based on system properties and the second on environment variables.

package com.gkatzioura.kafka.producer;

import java.util.Properties;

import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j

public class StrMessageProducer {

    private final Producer<String,String> producer;

    private final String topic;

    StrMessageProducer(Producer<String,String> producer, String topic) {

        this.producer = producer;

        this.topic = topic;

    }

    public void send(String message) {

        var producerRecord = new ProducerRecord<String,String>(topic,null, message);

        try {

            var metadata = producer.send(producerRecord).get();

            log.info("Submitted {}",metadata.offset());

        }

        catch (InterruptedException |ExecutionException e) {

            log.error("Could not send record",e);

        }

    }

    public static StrMessageProducer createFromSystemPros() {

        var kafkaProperties = new Properties();

        kafkaProperties.put("bootstrap.servers",System.getProperty("bootstrap.servers"));

        kafkaProperties.put("key.serializer",System.getProperty("key.serializer"));

        kafkaProperties.put("value.serialize",System.getProperty("value.serializer"));

        var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);

        return new MessageProducer(kafkaProducer, System.getProperty("main.topic"));

    }

    public static StrMessageProducer createFromEnv() {

        var kafkaProperties = new Properties();

        kafkaProperties.put("bootstrap.servers",System.getenv("BOOTSTRAP_SERVERS"));

        kafkaProperties.put("key.serializer",System.getenv("KEY_SERIALIZER"));

        kafkaProperties.put("value.serialize",System.getenv("VALUE_SERIALIZER"));

        var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);

        return new MessageProducer(kafkaProducer, System.getProperty("MAIN_TOPIC"));

    }

}

You already see the benefits. You have a clean class ready to use as it is and you have some factory methods for convenience. Eventually you can add more static factories, some of them might also have arguments, for example the topic.

Also we can go one step further when we want to have multiple classes of MessageProducers and we want to utilise an interface. So we are going to introduce the MessageProducer interface which our StrMessageProducer class will implement. Also we are going to put the static factories to the interface.

So this will be our interface with the static factories.

package com.gkatzioura.kafka.producer;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;

public interface MessageProducer {

    void send(String message);

    static MessageProducer createFromSystemPros() {

        var kafkaProperties = new Properties();

        kafkaProperties.put("bootstrap.servers",System.getProperty("bootstrap.servers"));

        kafkaProperties.put("key.serializer",System.getProperty("key.serializer"));

        kafkaProperties.put("value.serialize",System.getProperty("value.serializer"));

        var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);

        return new StrMessageProducer(kafkaProducer, System.getProperty("main.topic"));

    }

    static MessageProducer createFromEnv() {

        var kafkaProperties = new Properties();

        kafkaProperties.put("bootstrap.servers",System.getenv("BOOTSTRAP_SERVERS"));

        kafkaProperties.put("key.serializer",System.getenv("KEY_SERIALIZER"));

        kafkaProperties.put("value.serialize",System.getenv("VALUE_SERIALIZER"));

        var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);

        return new StrMessageProducer(kafkaProducer, System.getProperty("MAIN_TOPIC"));

    }

}

And this would be our new StrMessageProducer class.

package com.gkatzioura.kafka.producer;

import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j

public class StrMessageProducer implements MessageProducer {

    private final Producer<String,String> producer;

    private final String topic;

    StrMessageProducer(Producer<String,String> producer, String topic) {

        this.producer = producer;

        this.topic = topic;

    }

    @Override

    public void send(String message) {

        var producerRecord = new ProducerRecord<String,String>(topic,null, message);

        try {

            var metadata = producer.send(producerRecord).get();

            log.info("Submitted {}",metadata.offset());

        }

        catch (InterruptedException |ExecutionException e) {

            log.error("Could not send record",e);

        }

    }

}

Let’s check the benefits

◉ We can have various implementations of a MessageProducer class

◉ We can add as many factories we want that serve our purpose

◉ We can easily test the MessageProducer implementation by passing mocks to the constructors

◉ We keep our codebase cleaner