Finding Leader Broker in Kafka 0.8.0

With Kafka 0.8.0 the paradigm for storing data has changed. No longer can you connect to any broker and get access to the data for a topic and partition.

Starting with 0.8.0 Kafka has implemented replication, so now data is only on a subset of your Brokers (assuming you have more Brokers than count of replicas).

To figure out which Broker is the leader for your topic and partition, connect to any Broker and request the MetaData:

        kafka.javaapi.consumer.SimpleConsumer consumer  = new SimpleConsumer("mybroker01",9092,100000,64 * 1024, "test");
        List topics2 = new ArrayList();
        topics2.add("storm-anon");
        TopicMetadataRequest req = new TopicMetadataRequest(topics2);
        kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);
        List<kafka.javaapi.TopicMetadata> data3 =  resp.topicsMetadata();

The reply from Kafka contains information about every partition in the topic. Iterate through the response and find the partition you want, then call 'leader().host()' to figure out what Broker to connect to.

The following code shows all the partitions for a specific topic, including where each of the replicas resides.

        for (kafka.javaapi.TopicMetadata item : data3) {
           for (kafka.javaapi.PartitionMetadata part: item.partitionsMetadata() ) {
               String replicas = "";
               String isr = "";
               for (kafka.cluster.Broker replica: part.replicas() ) {
                   replicas += " " + replica.host();
               }
               for (kafka.cluster.Broker replica: part.isr() ) {
                   isr += " " + replica.host();
               }
              System.out.println( "Partition: " +   part.partitionId()  + ": " + part.leader().host() + " R:[ " + replicas + "] I:[" + isr + "]");
           }
        }

Note that you only have to do this if you are managing your own offsets and using the SimpleConsumer. Using Consumer Groups takes care of this automatically.

Also note that you don't need to do this on the Producer side, Kafka handles figuring out which partition is on which Broker for you.

Presenting Apache Kafka at AJUG

I'll be presenting about Apache Kafka at AJUG in March.

http://www.meetup.com/atlantajug/events/99878712/

I'll post up the slides after the presentation and may even have a video this time.

"Web Scale Architecture" study group

While reading about Netflix' downtime on Christmas Eve 2012 a thought entered my mind: I wonder what thinks about this?

From this thought came an idea for a study group at work. Our focus? Studying "web scale" architectures. Of course "web scale" can mean anything, so we decided to focus on things around best practices in Resiliency of systems, occasionally looking at cool algorithm implementations and 'how they built it' articles.

Of course there is no one book (and while we have a book budget, buying a bunch of books to read one chapter isn't a wise use of it!) we are instead looking at articles, blogs and videos.

Today was our first meeting (12-1 during lunch) and we had a very eclectic group of people in the room. Many software engineers, a few architects, a couple of DevOps, a couple of QA, a Security Engineer and a few product managers. We had a very good discussion about 'falling over' and the CircuitBreaker pattern.

Here is what we looked at.

Netflix blog about Resilient Systems:

http://techblog.netflix.com/2011/12/making-netflix-api-more-resilient.html

Circuit Breaker Pattern description from HubSpot:

http://dev.hubspot.com/blog/bid/64543/Building-a-Robust-System-Using-the-Circuit-Breaker-Pattern

Sample source code:

http://thatextramile.be/blog/2008/05/the-circuit-breaker

We of course went off a few tangents, but overall I really enjoyed discussing this with such a diverse group of people.

I am going to try to update this blog after each meeting in case you too are interested in learning about this stuff.

What does it mean to be a Senior Engineer?

Read a great blog post today about what it means to be a 'senior engineer'. Applicable to pretty much any discipline not just engineering. Same can be said for 'senior mechanics' or 'senior plumbers'.

http://www.kitchensoap.com/2012/10/25/on-being-a-senior-engineer/

This blog linked to a second blog with some great Fatherly advice for a son:

http://blog.stephenwyattbush.com/2012/04/07/dad-and-the-ten-commandments-of-egoless-programming

My favorite is definitely:

"Treat people who know less than you with respect, deference, and patience. Non-technical people who deal with developers on a regular basis almost universally hold the opinion that we are prima donnas at best and crybabies at worst. Don’t reinforce this stereotype with anger and impatience."

Finding Kafka JAR for building Java producers and consumers

File this one under 'why is this so hard?'.

I've started playing with Kafka and after getting a basic cluster up on 3 nodes, I started writing some code to produce my own events. It took me a good 30 minutes to figure out where the Kafka jar is to actually compile some code!

First step is to use the 'sbt' tool to build the installation. I ended up doing a couple of different targets to finally find the jar files.

./sbt update
./sbt package
./sbt package-all

Once all completed, I had to use 'find' to find the jar file to include in an IntelliJ project to compile and run a producer. The jar was finally found under C:\temp\kafka-0.7.2\core\target\scala_2.8.0\kafka-0.7.2.jar

Adding this jar to my IntelliJ project allowed me to compile and run a simple producer.

Setting up log4j rotating logger with MapR Hadoop

This one vexed me for several years and today I finally figured out how to do it!

We have a daemon that runs a Cascading job on request from a client. Since Cascading is build on top of Hadoop, the command to start the daemon looks like any other hadoop request.

The problem was getting the daemon to do a rotating appender in log4j. Since it is a daemon we want it to run lights out 24/7 so just redirecting stdin and stderr to a log file wasn't going to work.

So we wrote a script that sets the following environment variables before launching the daemon:


export HADOOP_LOG_DIR="/home/hadoop/mapr/hadoop_reporting/logs"
export HADOOP_LOGFILE="daemon.log"
export HADOOP_ROOT_LOGGER="INFO,console,DRFA"

The key change is the 'DRFA' line on the logger. You could ignore the first two and the logs would go to the default Hadoop location ($HADOOP_HOME/logs).

This is really useful since we can run different daemons, each writing to their own log files.

Note that this doesn't change the cluster-side logging. They all still go to where they are expected.

Finding "Number of Under-Replicated Blocks" in Hadoop

This one was bugging me for a long time. Even with the cluster idle, the Name Node summary would tell me there were a number of Under-Replicated Blocks in the system.

Turns out that all the Name Node problems we've been having were leaving 'temporary' files in HDFS and for whatever reason when we restarted the Name Node it wouldn't fix them.

I found them under: /log/hadoop/tmp/mapred/staging/<>/.staging/job_*

After confirming that the users weren't running active jobs, removing these directories via the command line reduced the number of blocks in the report and eventually all were cleared.

FYI our Name Node problems APPEAR to have been resolved in Cloudera CDH3 u3. Name Node has been up for 3 days now. Previously we were lucky if it lasted 48 hours.

Using the HDFS APIs without the hadoop main

I recently spent way too much time trying to run a simple Java application that uses the HDFS APIs to copy files into HDFS.

While using these APIs works great from within an application launched via the 'hadoop' command line, building one that is called from a Java main() was more challenging.

Why would I want to do this? Mainly because the process that looks for files, figures out where they should live in HDFS, selects from and updates an external database. Using Hibernate and Spring. But the Cascading application that uses the data in HDFS doesn't need Hibernate or Spring. So building a single Jar that supports both seemed overkill (and had its own issues with dependencies and final size of the jar)

First challenge was figuring out what parts of the hadoop shell file to duplicate, since not everything was needed or welcome. Turns out the information is in two places /bin/hadoop-config.sh and /conf/hadoop-env.sh

First try was to set $HADOOP_HOME and try '. $HADOOP_HOME/bin/hadoop-config.sh' in the script that calls my java main().

Except this stepped on HADOOP_HOME for some reason. Turns out the hadoop-config.sh file is playing some games with the path used to call the script to figure out what to set HADOOP_HOME to.

So calling '/opt/hadoop/bin/hadoop-config.sh' (note no '.' or shell variable!) set the path correctly.

then you can use '. $HADOOP_HOME/conf/hadoop-env.sh'

Finally, in your java classpath you need to set '$HADOOP_HOME/conf' before your classes.

For example: java -classpath .:$HADOOP_HOME/conf:/mycode ...

For the record, here is the error I was receiving that helped me find this:

java.io.IOException: Mkdirs failed to create /my_dir/my_file

Tired of hibernate lazy load errors on collections?

I have a love/hate relationship with Hibernate (Spring w/Hibernate to be fair). It makes some things very, very easy and others so difficult or convoluted.

For example, what should be trivial turns out to be a major pain:

Have an Entity object that contains two child collections. For example

@Entity
public class Client implements Serializable {
...

@OneToMany(mappedBy = "client")
private List users;

@OneToMany(mappedBy = "client")
private List sites;
...

}

Where User and Sites are also simple @Entity classes.

By default the access to these collections are Lazy Loaded. But when you want to access them both you run into problems. Setting both to 'eager' gets an error about too many buckets. Loading one Eager, then passing the object to a JSP for example, with get you the infamous:

org.hibernate.LazyInitializationException: failed to lazily initialize a collection of role: ...Client.organizations, no session or session was closed

(look on Google or StackOverflow for this error, there are no good solutions.)

So the hack/workaround? Create a custom Find method on the DAO and EXPLICITLY tell Hibernate to load the collection.

Note that using the collection directly in 'normal' code doesn't work since Spring/Hibernate/the compiler sees you aren't using the loaded collection and doesn't do the load.

However this works:

Hibernate.initialize(rtn.getOrganizations());

after your default 'find' method returns.

Details of how/why this works here.

Even more bizarre, when debugging this, setting a breakpoint in the 'find' method everything works correctly. No breakpoint and you get the exception.

(Thanks to Scott Mitchell for his help on pointing me to this solution.)

Logging in Groovy shouldn't be this hard

Spent about 30 minutes this morning doing what should have been easy: logging from within my application.

Groovy includes/wraps Log4j so I thought it would be easy. All the documentation I found suggested it would be easy.

However all the examples left off one key thing: Defining the 'root' logger.

So, in your Config.groovy, find the log4j section and add/uncomment:

appenders {
console name:'stdout', layout:pattern(conversionPattern: '%c{2} %m%n')
}

Then add below the standard error and warn items:

root {
info 'console'
}

Now in your code you can add 'log.info 'blah blah' and it will appear on the console. The 'appenders' section is where you can add your rolling file loggers for production.

Here is what mine looks like:

// log4j configuration
log4j = {
// Example of changing the log pattern for the default console
// appender:
//
appenders {
console name:'stdout', layout:pattern(conversionPattern: '%c{2} %m%n')
}


error 'org.codehaus.groovy.grails.web.servlet', // controllers
'org.codehaus.groovy.grails.web.pages', // GSP
'org.codehaus.groovy.grails.web.sitemesh', // layouts
'org.codehaus.groovy.grails.web.mapping.filter', // URL mapping
'org.codehaus.groovy.grails.web.mapping', // URL mapping
'org.codehaus.groovy.grails.commons', // core / classloading
'org.codehaus.groovy.grails.plugins', // plugins
'org.codehaus.groovy.grails.orm.hibernate', // hibernate integration
'org.springframework',
'org.hibernate',
'net.sf.ehcache.hibernate'

warn 'org.mortbay.log'

root {
info 'console'
}