Introduction

Since a few months I'm diving into Apache Kafka. I've always been fascinated by queuing mechanisms.  And Apache Kafka nowadays is the most modern alternative. Lately I did a presentation on an introduction to Apache Kafka:




But now I'm investigating what I can do with it. Since Weblogic is one of my focus areas, I wanted to explore how I can embed Kafka into Weblogic.

I reasoned that when I want to use Kafka with a current customer, the administrators have to install kafka (eg. unzip the Confluent distribution), on a separate virtual server.
By default the distribution comes with startup and shutdown scripts. The administrators should use those, or create their own, and startup the Kafka and Zookeeper services. And of course keep those up-and-running.

I figured that when I would be able to start the services as a thread under a Weblogic server, no additional infra structure is needed. Also starting the Weblogic server would start the Kafka services as well.

Kafka needs a ZooKeeper service. You can see the ZooKeeper as a directory service for a Kafka infrastructure. Slightly comparable to an AdminServer in Weblogic. So it would make sense, as I see it, to start the ZooKeeper with the AdminServer. The Kafka Servers can be started as part of the Weblogic Managed Server(s)

Weblogic has a mechanism to do initializations and finalizations, using startup and shutdown classes, see these documentation. From there the ZooKeeper and KafkaServers can be started.

So I had to figure out how to start those from Java. Let's start with the ZooKeeper.
I put my sources on GitHub, so you can review them. But keep in mind that they're still under construction.

Starting a ZooKeeper

My starting point was this question on StackOverflow, that handles starting a ZooKeeperServer in Java, based on the ZooKeeperServerMain.java class. It was quite promising and soon I had a first version of my startup class working. Quite simple really. But, since I also want to be able to shut it down, I soon ran into some restrictions. Some methods and attributes I needed were protected and only reachable from the same package, for instance. I wasn't quite pleased with the implementation. Digging a bit further I ran into the source of that class over here. I decided to take that class, study it and based on that knowledge implement my own class.

I created a ZooKeeperObserver class, and transformed the public void runFromConfig(ServerConfig config) method from ZooKeeperServerMain.java class, into a public void runFromProperties(ZooKeeperProperties zkProperties) method.

It takes in a properties object, that is interpretted and used to start the ZooKeeper.

Zookeeper Properties

To keep things transparent and simple, I created a PropertiesFactory class that provides a method to read the zookeeper.properties from the class path (therefor we should add the /etc/kafka folder to it).
I also created an own Properties class extending java.util.Properties to add a few property getter methods, like getting an int value and defaulting a property based on an other property.

Lastly, I created the ZooKeeperProperties bean, to interpret the relevant ZooKeeper properties, from a read Properties object.

The relevant properties are:

Property	Meaning	Default
dataDir	The location where ZooKeeper will store the in-memory database snapshots and, unless specified otherwise, the transaction log of updates to the database.	/tmp/zookeeper
dataLogDir	This option will direct the machine to write the transaction log to the dataLogDir rather than the dataDir.	dataDir
clientPort	The port to listen for client connections; that is, the port that clients attempt to connect to.	2181
clientPortAddress	The address (ipv4, ipv6 or hostname) to listen for client connections; that is, the address that clients attempt to connect to.	Empty: every NIC in the server host.
maxClientCnxns	Limits the number of concurrent connections (at the socket level) that a single client, identified by IP address.	0: disabled, since this is a non-production config.
tickTime	The length of a single tick, which is the basic time unit used by ZooKeeper, as measured in milliseconds.	ZooKeeperServer.DEFAULT_TICK_TIME
minSessionTimeout	The minimum session timeout in milliseconds that the server will allow the client to negotiate. Defaults to 2 times the tickTime.	-1: Disabled
maxSessionTimeout	The maximum session timeout in milliseconds that the server will allow the client to negotiate. Defaults to 20 times the tickTime.	-1: Disabled

Only the properties dataDir, clientPort and maxClientCnxns are set explicitly in the zookeeper.properties file. See the Zookeeper Administration docs for more info (apparently Zookeeper is created/invented in the Hadoop project).

Run from Properties

The runFromProperties is the one that actually starts a ZooKeeperServer instance:

    /**
     * Run from ZooKeeperProperties .
     * @param zkProperties ZooKeeperProperties to use.
     * @throws IOException
     */
    public void runFromProperties(ZooKeeperProperties zkProperties) throws IOException {
        final String methodName = "runFromProperties";
        log.start(methodName);
        log.info(methodName, "Starting server");
        FileTxnSnapLog txnLog = null;
        try {
            // Note that this thread isn't going to be doing anything else,
            // so rather than spawning another thread, we will just call
            // run() in this thread.
            // create a file logger url from the command line args
            ZooKeeperServer zkServer = new ZooKeeperServer();

            txnLog = new FileTxnSnapLog(new File(zkProperties.getDataLogDir()), new File(zkProperties.getDataDir()));
            zkServer.setTxnLogFactory(txnLog);
            zkServer.setTickTime(zkProperties.getTickTime());
            zkServer.setMinSessionTimeout(zkProperties.getMinSessionTimeout());
            zkServer.setMaxSessionTimeout(zkProperties.getMaxSessionTimeout());
            setZooKeeperServer(zkServer);

            cnxnFactory = ServerCnxnFactory.createFactory();
            log.debug(methodName, "Create Server Connection Factory");
            log.debug(methodName, "Server Tick Time: " + zkServer.getTickTime());
            log.debug(methodName, "ClientPortAddress: " + zkProperties.getClientPortAddress());
            log.debug(methodName, "Max Client Connections: " + zkProperties.getMaxClientCnxns());
            cnxnFactory.configure(zkProperties.getClientPortAddress(), zkProperties.getMaxClientCnxns());
            log.debug(methodName, "Startup Server Connection Factory");
            cnxnFactory.startup(zkServer);
            cnxnFactory.join();
            if (zkServer.isRunning()) {
                zkServer.shutdown();
            }
        } catch (InterruptedException e) {
            // warn, but generally this is ok
            log.warn(methodName, "Server interrupted", e);
        } finally {
            if (txnLog != null) {
                txnLog.close();
            }
        }
        log.end(methodName);
}

Here you see that a ZooKeeperProperties is passed. A FileTxnSnapLog is initialized for the dataDir and dataLogDir. A ZooKeeperServer is instantiated, and the particular properties are set. Then a ServerCnxnFactory is created (as a class attribute for later use). The connection factory is used to startup the ZooKeeperServer.  Actually, at that point the control is handed over to the ZooKeeperServer. So, you want to have this done in a separate thread.

Observing the Observable

Now, you might think: What is it with the name ZooKeeperObserver? Earlier, I named it EmbeddedZooKeeperServer. But I found that name long and not nice. I found it funny that Observer has the word Server in it.

As mentioned in the previous section, when starting up the ConnectionFactory/ZookeeperServer, the control is handed over. The method is not left, until the ZooKeeperServer stops running.

I therefor want (as in many implementations) that the ZooKeeperServer, runs in a seperate thread, that I can control. That is, I want to be able to send a shutdown signal to it. For that I found the Observer pattern suitable. In this pattern, the Observable or Subject maintains a list of Observers that can be notified about an update in the Observable. To do so, the Observable extends the java.util.Observable class. And the Observer implements the java.util.Observer and Runnable interfaces.

How does it work? Let's go through the applicable methods.

Start and Add a ZooKeeper

The Observable is implemented by ZooKeeperDriver. In it we'll find a method start():

    public void start() {
        final String methodName = "start";
        log.start(methodName);
        addZooKeeper();
        log.end(methodName);
}

That's not too exiting, but it calls the method addZooKeeper():

    public void addZooKeeper() {
        final String methodName = "addZooKeeper";
        log.start(methodName);
        try {
            ZooKeeperProperties zkProperties = PropertiesFactory.getZKProperties();
            ZooKeeperObserver zooKeeperServer = new ZooKeeperObserver(this, zkProperties);
            Thread newZooKeeperThread = new Thread(zooKeeperServer);
            zooKeeperServer.setMyThread(newZooKeeperThread);
            newZooKeeperThread.start();
        } catch (IOException e) {
            log.error(methodName, "ZooKeeper Failed", e);
        }
        log.end(methodName);
    }

Here you see that the ZooKeeperProperties are fetched and a new ZooKeeperObserver is instantiated, using a reference to the ZooKeeperDriver object and the ZooKeeperProperties. Since the ZooKeeperObserver is a Runnable we can add it to a new Thread. That thread is also set to the ZooKeeperObserver so that it has a hold of it's own thread, when that come in handy.
And then the new thread is started.

Instantiate the ZooKeeperObserver

In the previous section, we saw that the ZooKeeperObserver is instantiated using a reference to the ZooKeeperDriver object. Let's see how it looks like:

    public ZooKeeperObserver(Observable zooKeeperDriver, ZooKeeperProperties zkProperties) {
        super();
        final String methodName="ZooKeeperObserver(Observable, ZooKeeperProperties)";
        log.start(methodName);
        this.setZkProperties(zkProperties);
        if (zooKeeperDriver instanceof ZooKeeperDriver) {
            log.info(methodName, "Add observer "+this.getClass().getName()+" to observable "+zooKeeperDriver.getClass().getName());
            setZooKeeperDriver((ZooKeeperDriver) zooKeeperDriver);
            zooKeeperDriver.addObserver(this);
        }
        log.end(methodName);
}

The ZooKeeperProperties are set. And then it checks if the Observable that is passed is indeed a ZooKeeperDriver. The ZooKeeperDriver is also set, and then the ZooKeeperObserver object is added as an Observer to the ZooKeeperDriver using the addObserver(this) method. This method is part of the java.util.Observable object that is extended. It adds the ZooKeeperObserver to a list, that is used to send the update signal to every instance on the list.

Run the ZooKeeperObserver

The ZooKeeperObserver is a Runnable so the run() method is implemented:

    public void run() {
        final String methodName = "run";
        log.start(methodName);
        try {
            runFromProperties(getZkProperties());
        } catch (IOException ioe) {
            log.error(methodName, "Run failed!", ioe);
        }
        log.end(methodName);
}

It calls the  runFromProperties(), that is explained earlier.

Shutdown

The ZooKeeperDriver has a shutdown() method:

    public void shutdown() {
        final String methodName = "shutdown";
        log.start(methodName);
        setShutdownZooKeepers(true);
        log.info(methodName, "Notify Observers to shutdown!");
        this.setChanged();
        this.notifyObservers();
        log.end(methodName);
}

It sets the shutdownZooKeepers indicator to true. This is an attribute that indicates what has been updated. In a more complex Observer pattern more kinds of updates can occur. So, you need to indicate what drove the update.
The most interesting statement is the call to the notifyObservers() method. It will call the implemeneted update() on every Observer in the list.

I implemented this earlier in another situation, a few years ago. And I reused it. But at first it did not work. I found that, apparently changed in Java 7 or 8, I had to add a call to the setChanged() method. The notification to the Observers only works after that call.

As said, notifyObservers() calls the update() method in the Observer:

    public void update(Observable o, Object arg) {
        final String methodName = "update(Observable,Object)";
        log.start(methodName);
        log.info(methodName, getMyThread().getName() + " - Got status update from Observable!");
        ZooKeeperDriver zkDriver = getZooKeeperDriver();
        if (zkDriver.isShutdownZooKeepers()) {
            log.info(methodName, getMyThread().getName() + " - Apparently I´ve got to shutdown myself!");
            shutdown();
        } else {
            log.info(methodName, getMyThread().getName() + " - Don't know what to do with this status update!");
        }
        log.end(methodName);
}

And this one actually checks in the ZooKeeperDriver if the change is because of the shutDownZooKeepers indicator.
If so, it calls it's own shutdown() method. If not, then the update is ignored. The shutdown does the following:

        final String methodName = "shutdown";
        log.start(methodName);
        log.info(methodName,"Let me shutdown "+myThread.getName());
        ZooKeeperServer zkServer = getZooKeeperServer();
        ServerCnxnFactory cnxnFactory = getCnxnFactory();
        cnxnFactory.shutdown();
        if (zkServer.isRunning()) {
            zkServer.shutdown();
        }
        log.end(methodName);
}

It gets the Connection factory and sends a shutdown() signal to it. if the ZooKeeper is still running (it shouldn't be), then it gets a shutdown() signal also.

Start and Shutdown

In the end you need to create an instance of the ZooKeeperDriver and save it into a static variable. Then you can call the start() method and later get the object again from the static variable, to call the shutdown() method.

Conclusion

This may look a quite complex to you, to start a server. But, again, I want to be able to embed the Kafka infrastructure in an other system, in my situation Weblogic. This method I'll use to do the same for the Kafka Servers. I'll write about that in a follow-up article. And then, I'll create a set of startup and shutdown classes for Weblogic.

It was fun to implement the Observer pattern again. But, when I encountered that the notifyObserver method did not work as expected at first, searching for a solution, I found that it is deprecated in Java 9. It will still work, but apparently people found that it has it's limitations and a better way of implementing it is developed.