Web problems and solutions.

Elasticsearch data migration

2016-06-25T22:23:00.000-07:00

Lately, I have been working on migrating from elasticsearch 0.19 to 1.7 version. One of the tasks involves migrating data from old to new cluster. In this post, I write about a tool for migrating the data over to the new version.

Firstly, you cannot move the index files directly, attempting results in following error in 1.7 ES:

Error injecting constructor, org.elasticsearch.ElasticsearchIllegalStateException: Could not find a state file to recover from among

Elasticsearch 0.19 version is ancient and not compatible for a direct migration. Hence the only solution is to read it from the old elasticsearch cluster and write to new elastic.

Followed a simple guideline while coming up for the tool:

Primary goal: "Just move the data over to new version!"
Nice to haves: "Monitor its progress"

A bare bone version of it is here in github capturing the essence of this migration. Primary goal is achieved by using Nodejs elasticsearch client library where given a index name, migrates based on created field in the document. A assumption being made that a document with timestamp is present but the solution can be used for any increasing sequence. Secondary goal is achieved by a in memory map which stores and updates the index status.

Checkout the README on github to set it up locally.

Whats pending:
A nice user experience.

Service continuity using Zookeeper

2015-08-28T10:39:00.000-07:00

In service oriented architecture where one service is dependent on others, the challenges one would face is how to upgrade one service without coming to a grinding halt.

Consider two servers (essentially both are services providing a broader service) A and B, with A depending on B. Lets assume for simplicity service B exposes REST api's to be consumed by service A.

Old way: A and B running on the same physical machine. This setup is replicated on number of machines.

Below are some of the concerns.

How to guarantee service continuity during software upgrade? Lets say B needs to be upgraded, should be halt service provided A? This maybe acceptable if A's service is not that critical and a downtime is acceptable.
How to avoid single point of failure? Lets say B stops responding, what should A do? Retry later? What should be the interval of retry?

Why it didnt work for us?
We had a chain of services where customer downtime was not acceptable. The replication on different machines handled the downtime part in a hacky fashion where customer facing service was load balanced but didn't address the single point of failure issue. Also this was not efficient utilization of resource. Consider a case where we need to scale up, in that case, we would have to add the entire stack.

New solution: Store B server state information in Zookeeper and let A discover B through Zookeeper. Curator's service discovery framework has a nice library which can be used for the same. Store the hostname, ip address, port (state information) when starting the service and remove this information during service stop.

Next steps: Will see if I can hack together a bare bones code and post a link here.

A little caution with "Mysql's Timestamp"

2012-10-01T16:54:00.002-07:00

In this post, I would like to bookkeep a interesting fact on Timestamp.

Fundamentally, Mysql Timestamp is nothing but seconds from epoch and differs from datetime as it stores the date time in UTC and abstracts the date time conversion to system's timezone. Also it supports auto-initialization ('CURRENT_TIMESTAMP or 0000-00-00 00:00:00') and auto-updates (CURRENT_TIMESTAMP).

Now the thing which one has to understand is passing "null" references to Timestamp column coupled with not null references and hence want to emphasize it in this post.

Consider a case where we create a table with definition of modified column as follows:
create table demo_null_passing_to_timestamp(...., modified timestamp not null default '0000-00-00 00:00:00')

The behavior for "modified" column will be as one expects if there is no value passed in to modified column, it will use the default value of 0000-00-00 00:00:00, but the behavior changes when null is passed in this case because of the "not null" constraint. In this case, it uses the value from current timestamp. The only way is to allow null value for this modified column.

Try passing null value to such a column to verify this behavior...

Configuration management Using ZooKeeper.

2012-07-20T18:28:00.001-07:00

Consider a case where dynamic reload of properties is required. An application can leverage ZooKeeper's service for providing such a capability.

The concept is that of a Publish Subscribe model but spanned across mulitple VMs (either on same machine or different).
The solution is of two folds:
1) First one being a file change listener. This piece of code is responsible for detecting any change and updating the data stored in Zookeeper.
2) Set a watch from the client on this data node being changed.

A quick demo should help clarify the concept.
1) zk-file-change-watcher.jar (executes java class ZkFileChangeWatcher): All it does is creates node in Zookeeper data tree "/CONFIG/DYNAMIC-FILE" if it doesnt exist and sets a watch. If this node changes, ZooKeeper service will notify this client of the change which it takes action (prints to the screen). As seen in screenshot (Denoted by "Waiting for changes" messages in terminal 1 and 2 on L.H.S.)

2) fs-file-change-watcher.jar (executes java class FSFileChangeWatcher): This creates a file listener and checks (polling) every 5 seconds for file changes by checking its timestamp. If the file has changed, it reads data from the file and pushes that change to data node: /CONFIG/DYNAMIC-FILE created by #1. Note when this jar runs for the first time, it doesn't have last updated timestamp, hence it displays message "Last modified timestamp changed" and pushes latest data to the node. Terminal on the R.H.S. displays this execution of this jar.

3) When the file (dynamic-file.txt) changes, both the listeners created in #1 and #2 act accordingly and update is displayed to the clients.
Below screenshot displays just before updating the file (4 is being added to the file) in vi editor. Displayed in bottom left terminal.

Below screenshot confirms the update received at the client in terminals #1 and #2. Terminal on right hand side displays File time stamp change message.

Zookeeper: Are you the HERO we were looking for?

2012-07-15T14:16:00.000-07:00

Recently I have been investigating on feasibility of using Zookeeper (super excited about it!!! :) ) for a existing platform.

Existing Architecture:
Briefly, the module does a bunch of stuff from

Registering itself when instance is started.
Work delegation among other instances through rpc
Acquiring lock for doing work (note due to distributed nature, is it stored outside of VM)

I agree all this works just fine! Why is there a need to investigate alternatives?

Concerns:

Synchronous remote method invocation: When one instance delegates work to another, the request goes through Spring's HttpInvoker which under the hood uses apache commons HttpClient. Since the underlying methods are either a GET/POST, I believe the remote method invocation results in synchronous calls effectively blocked until results are retrieved. The remote calls are expensive ( atleast 4-5 times than a normal method invocation). More details can be found in the paper. Currently it works because there are two production instances invoking each other and it has to deal with transactions in the range 17-20k a day. Looking 6 months back we have doubled our processing volume, continuing that trend, in a years time we might be at 80k transactions a day. Will our current solution be sufficient for scale?
Fail over / High availability: Both instances are parallel running and both try to do the same work. That is both are configured to run in Master mode. The way to do any work to avoid duplicate is by acquiring a shared lock. Thus High Availability is achieved through replication but it is not the right way to do it. In a ideal situation, they should be configured as Master/Slave with the ability to take over by Slave in case if Master fails.
Locking: Locks are granulated based on the type of work and are also classified into blocking/non-blocking locks. Concerning part is they are acquired and released everytime when a work is done. More instances, more of this, which will increase the load on database.
Lastly, this allocation part makes the system highly coupled and confusing.

It seems, Zookeeper can be used to provide Leader Election/ group membership, Distributed locking and is highly available. The original paper and recipe's page gives a good overview of what Apache Zookeeper is.

In my next blog, I demonstrate few features which can be used to correct above fallacies like Leader Election, Configuration management, Distributed Locking, Group Membership.

Zookeeper: Leader Election and Group membership.

2012-07-15T14:15:00.001-07:00

Its been less than a week since I started playing with Zookeeper to understand how can it be used for our existing platform and in this post I would like to document its Leader Election feature.

Used the algorithm pointed out in the recipe page. Few key things to note:

ZooKeeper is analogous to a file system where each Node can be thought of either a DIRECTORY (intermediate nodes which contains file or subdirectory) or FILE (leaf nodes) with root node denoted as unix style "/".
Make the nodes EPHEMERAL and SEQUENTIAL. This can be easily achieved by passing CreateMode.EPHEMERAL_SEQUENTIAL while create a node using create() method.
Implement Watcher Interface. Think of Watcher objects as Listeners which register with ZooKeeper for listening on an event.
Following diagram should help understand the concept visually.

The algorithm is self explanatory along with commented code.

Lets move to a working example. Compiled the code into jar and ran as independent processes. The bottom right window in the screenshot is the zookeeper server's view, connected through its command line interface (zkCli.sh)

1) Initial state:

2) When first node connects (First node, NODE-0000000060 connects to Zookeeper server)

3) Similarly, second and third instance connect to zookeeper server. (Notice the list commands issued in the bottom right window confirms that they have registered and NODE-0000000061 is following 60, NODE-0000000062 is following 61)

4) Leader, NODE-0000000060 is killed. Notice, how the NODE-0000000061 follower takes up Leader's responsibility.

5) Resurrecting the process in Top left window, notice it now becomes a follower, NODE-0000000064 and starts following node, NODE-0000000062.

6) Kill a intermediate follower (process which is not a leader nor follower but an in between node which forms a logical Watcher link. Note reelection should be initiated to fix the broken link so that Leader responsibilities can be passed to the last remaining node if all previous nodes happen to crash. In the following screenshot the intermediate node, NODE-00000062 is between the leader, NODE-00000061 and node, NODE-00000064. If the intermediate node crashes, the node, NODE-00000064 starts following the leader, NODE-00000061. Also Node, NODE-00000064 becomes leader when leader, NODE-00000064 crashes.

Build a jar (leader-election.jar) using the source code.

Thus leader election and group management can be easily implemented using ZooKeeper. Next moving on to Distributed Locking using ZooKeeper.

Conditional Probability

2012-07-08T21:52:00.002-07:00

I was trying to rekindle my interest over the weekend in Text Classification and hence wanted to revisit my probability skills. I would like to document it so in future I don't have to spend time understanding the basics.

Simply stated, conditional probability is given an event "A" what is the chance another event "B" can occur. As with anything, this statement can be further explained in depth with a example.

Consider 5 marbles (2 red, 3 blue), what it the probability of having 2 blue marbles taken from a bag in consecutive attempts. Key thing to understand is the second event is dependent on the first. The first event should draw a blue marble, followed by the second marble drawn to be blue.

So the chance that first event of drawing blue ball is 2/5. Now, the remaining marbles in the bag are 4, hence the chance of drawing another blue ball is 1/4. Conditional probability is multiplying the dependent events, hence we get: (2/5) * (1/4) = (1/10).

Thus the chance of having two blue balls 10%.

More at:
1) http://www.mathsisfun.com/data/probability-events-conditional.html
2) Shared Birthday puzzle.

How is this related to Document Text Classification?
What is the chance that a given document belongs to a classification (good/bad/unknown)?
I have been progressing through Programming Collective Intelligence and in Chapter 6, Document Classification has a nice description on how to find one!

Avoiding WebappClassLoader memory leaks.

2012-07-05T18:15:00.000-07:00

Recently, I have been working to solve memory leak issues of our Integration web application and would like to document my findings.

Issue is WebappClassLoader object is not de-referenced when the application context is destroyed and hence cannot be garbage collected and as a result all objects which are referenced from WebappClassLoader reside in memory (to be precise in PermGenSpace part of the heap)

Following are interesting links about this issue:
1) Frank Kieviet's blog.
2) A Nice tutorial on diagnosing memory leak.
3) Creating memory leaks in java.
4) Class loader in deep.
5) As always, Java language specification is a great guide.

It turns out, the web application had a couple of issues:
1) Static final object reference: Static means only one object can persist in memory and making that object final makes it even harder to make it available for garbage collection by "nulling the reference".
Issue: The application sneaky created a reference to this Log4j object through FileWatchdog (instantiated through PropertyConfigurer). Because these object references are maintained in HashTable (object "ht") inside Log4j library's Hierarchy class restarting the web application didn't clear this reference as it was also referenced from a thread which was never stopped. More details at #4 below.
Solution: Removed static keyword on the object because there was only one object of the class ever to exist. It worked for us, it might not for you, take a look at the Log4jConfigListener (define it in web.xml before other listeners) for another solution. This manages Log objects when the during the life cycle of the web application.

2) There were threads which were started during web application startup but never stopped. This prevented from WebappClassLoader garbage collection. The code was not properly architected, even the InterruptedException was swallowed by the thread and continued processing indefinately.
Solution: Making the thread response to interrupts and return on InterruptedException.

3) A module in the web application fires jobs based on cron expression at regular intervals. The SchedulerFactoryBean.destroy() was never called when the context was destroyed.
Solution: Modified the spring bean with destroy-method="destroy" in the application context spring bean definition. Note, this might not work for older version of spring as discussed here. In this case, use the solution discussed of writing a your servlet listener which destroys the quartz classes when the app context is destroyed.

4) FileWatchDog Thread in Log4j: This is most serious of all issues because there is no solution yet to this. This thread is created when dynamic loading of log4j.properties without having to re-starting web container (in our case, it is tomcat), by PropertyConfigurer.configureAndWatch. (note, every invocation of this method creates a new thread)
Issue: The code doesn't take into account interruption (InterruptedException is swallowed), no way to shutdown this thread.
From stackoverflow, some solutions can be concluded but they wont work unless VM shuts down. The reason being, there is no way to get to this thread (i.e. PropertyConfigurer does not have a reference of this instantiated thread)

Using LogManager.shutdown(), this method does shutdown every Log appenders and Category associated with the Root but not FileWatchDog thread.
Defining Log4jConfigListener in web.xml doesnt help because when the context is destroyed, internally it calls LogManager.shutdown() (through Log4jWebConfigurer.shutdownLogging)

Solution

I have to admit that not using the dynamic reloading of log files in tomcat is the easiest!
More hacky way, the thread FileWatchdog, repeatedly executes the method: checkAndConfigure(). Below is the code snippet:

protected
  void checkAndConfigure() {
    boolean fileExists;
    try {
      fileExists = file.exists();
    } catch(SecurityException  e) {
      LogLog.warn("Was not allowed to read check file existance, file:["+
          filename+"].");
      interrupted = true; // there is no point in continuing
      return;
    }

    if(fileExists) {
      long l = file.lastModified(); // this can also throw a SecurityException
      if(l > lastModif) {           // however, if we reached this point this
    lastModif = l;              // is very unlikely.
    doOnChange();
    warnedAlready = false;
      }
    } else {
      if(!warnedAlready) {
    LogLog.debug("["+filename+"] does not exist.");
    warnedAlready = true;
      }
    }
  }

If the file permission just before context destroyed (maybe define a custom listener) was made read only, the method checkAndConfigure will throw a SecurityException which causes the boolean "interrupted" to set and hence will break the loop. Dont forget to reset the file persmission on context initialization.

We went ahead with the first solution as dynamic reload didn't add much value but hopefully the second solution might help someone...

Jvm translations which are easy to overlook.

2012-05-29T14:32:00.001-07:00

Recently I came across a code which gave a false sense of synchronization guarantee. Also this code reminded me how easy it is to forget about JVM byte code translations.

Issue was a synchronized list was being populated in one thread and accessed through another. For simplicity reasons, following code could be treated as a similar example (few threads updates the list and others iterate over it):

List<String> list = new ArrayList<String>();
list = Collections.synchronizedList(list);
list.add("a");
list.add("b");
list.add("c");
list.add("d");
list.add("e");
for (String temp: list) {
   System.out.println("value:"+temp);
}

Do you think above code would give synchronization guarantee when iterating in the loop? No, it wont. I am not doubting the synchronization policy on list operations (add, get, delete, etc) but over its element iteration. To prove the case, get a decompiler (eg: jad) and from the class file reconstruct the java file. Below is the snapshot of the same.

Above snapshot confirms that there is no synchronization taking place during iteration. Changed the code to following to enforce synchronization (alternatively can use JAVA 6, concurrent variant CopyOnWriteArrayList):

List<String> list = new ArrayList<String>();
list = Collections.synchronizedList(list);
list.add("a");
list.add("b");
list.add("c");
list.add("d");
list.add("e");
synchronized (list) {
  for (String temp: list) {
     System.out.println("value:"+temp);
  }
}

Below snapshot confirms (decompiled version) the enforcement (check for monitorenter):

Side effect would have caused list corruption (update by one thread while in use by another thread. Also this issue can also be confirmed through test case (Iterators are fail fast and with an unlucky timing would result in ConcurrentModificationException)

Few noteworthy things besides the synchronization issue:

1) "For-each Loop" construct introduced in Java 5 gets translated into Iterator.
2) String concatenation using "+" operator gets translated to "new StringBuilder.append()...append().toString()".

Using maven to configure a hadoop project.

2012-05-10T09:59:00.000-07:00

In this post, I plan to describe how and benefits of using maven to configure a hadoop project.

First comes the benefits: There are many described here, but it benefited me in writing testcode, creating and deploying jars using one single command.

Now the how part: The first thing is to be execute the following command which creates the project directory (along with package structure and pom.xml)

mvn archetype:generate -DgroupId=com.mycompany.example1 -DartifactId=example1 -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

Notice a example1 project would be created (which matches articfactId) and a package hierarchy inside src/com/mycompany/example1 will be present (this matches the groupId)

Now open the pom.xml inside example1 and the following inside <dependencies></dependencies> tags.

<dependency>
  <groupId>junit</groupId>
  <artifactId>junit</artifactId>
  <version>3.8.1</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-core</artifactId>
  <version>1.0.1</version>
</dependency>
<dependency>
  <groupId>org.jmock</groupId>
  <artifactId>jmock-junit4</artifactId>
  <version>2.5.1</version>
</dependency>
<dependency>
  <groupId>org.jmock</groupId>
  <artifactId>jmock-legacy</artifactId>
  <version>2.5.1</version>
</dependency>

Specifying above dependencies inside POM.xml (project object model) will grab the required jars (in this case, will pull hadoop-core 1.0.1, junit 3.8.1, jmock 2.5.1 respectively from the maven repository)

Now we are all set, using your favorite IDE, start your coding!

Check the directory structure using maven dependency tree.

mvn dependency:tree

The above project should be as follows

[INFO] [dependency:tree {execution: default-cli}]
[INFO] com.xoom.example3:example3:jar:1.0-SNAPSHOT
[INFO] +- junit:junit:jar:3.8.1:test (scope not updated to compile)
[INFO] +- org.apache.hadoop:hadoop-core:jar:1.0.1:compile
[INFO] |  +- commons-cli:commons-cli:jar:1.2:compile
[INFO] |  +- xmlenc:xmlenc:jar:0.52:compile
[INFO] |  +- commons-httpclient:commons-httpclient:jar:3.0.1:compile
[INFO] |  |  \- commons-logging:commons-logging:jar:1.0.3:compile
[INFO] |  +- commons-codec:commons-codec:jar:1.4:compile
[INFO] |  +- org.apache.commons:commons-math:jar:2.1:compile
[INFO] |  +- commons-configuration:commons-configuration:jar:1.6:compile
[INFO] |  |  +- commons-collections:commons-collections:jar:3.2.1:compile
[INFO] |  |  +- commons-lang:commons-lang:jar:2.4:compile
[INFO] |  |  +- commons-digester:commons-digester:jar:1.8:compile
[INFO] |  |  |  \- commons-beanutils:commons-beanutils:jar:1.7.0:compile
[INFO] |  |  \- commons-beanutils:commons-beanutils-core:jar:1.8.0:compile
[INFO] |  +- commons-net:commons-net:jar:1.4.1:compile
[INFO] |  +- org.mortbay.jetty:jetty:jar:6.1.26:compile
[INFO] |  |  \- org.mortbay.jetty:servlet-api:jar:2.5-20081211:compile
[INFO] |  +- org.mortbay.jetty:jetty-util:jar:6.1.26:compile
[INFO] |  +- tomcat:jasper-runtime:jar:5.5.12:compile
[INFO] |  +- tomcat:jasper-compiler:jar:5.5.12:compile
[INFO] |  +- org.mortbay.jetty:jsp-api-2.1:jar:6.1.14:compile
[INFO] |  |  \- org.mortbay.jetty:servlet-api-2.5:jar:6.1.14:compile
[INFO] |  +- org.mortbay.jetty:jsp-2.1:jar:6.1.14:compile
[INFO] |  |  \- ant:ant:jar:1.6.5:compile
[INFO] |  +- commons-el:commons-el:jar:1.0:compile
[INFO] |  +- net.java.dev.jets3t:jets3t:jar:0.7.1:compile
[INFO] |  +- net.sf.kosmosfs:kfs:jar:0.3:compile
[INFO] |  +- hsqldb:hsqldb:jar:1.8.0.10:compile
[INFO] |  +- oro:oro:jar:2.0.8:compile
[INFO] |  +- org.eclipse.jdt:core:jar:3.1.1:compile
[INFO] |  \- org.codehaus.jackson:jackson-mapper-asl:jar:1.0.1:compile
[INFO] |     \- org.codehaus.jackson:jackson-core-asl:jar:1.0.1:compile
[INFO] +- org.jmock:jmock-junit4:jar:2.5.1:compile
[INFO] |  +- org.jmock:jmock:jar:2.5.1:compile
[INFO] |  |  +- org.hamcrest:hamcrest-core:jar:1.1:compile
[INFO] |  |  \- org.hamcrest:hamcrest-library:jar:1.1:compile
[INFO] |  \- junit:junit-dep:jar:4.4:compile
[INFO] \- org.jmock:jmock-legacy:jar:2.5.1:compile
[INFO]    +- org.objenesis:objenesis:jar:1.0:compile
[INFO]    \- cglib:cglib-nodep:jar:2.1_3:compile

Unit testing Mapper.

2012-05-09T13:58:00.000-07:00

In this post, I am writing a simple unit test using JMock for mapper mentioned previously. Briefly, the test case is validating the key and value collected by the context object. In this test case, context object is mocked and hence expectations can be set on the object. Below code is self explanatory.

package com.mycompany.example3;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.jmock.Expectations;
import org.jmock.Mockery;
import org.jmock.integration.junit4.JUnit4Mockery;
import org.jmock.lib.legacy.ClassImposteriser;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class MapClassTest {

    private MapClass mapper;
    private Text text;
    private Mapper.Context context;
    private Mockery mockery = new JUnit4Mockery() {{
        setImposteriser(ClassImposteriser.INSTANCE);
    }};

    @Before
    public void setUp() throws Exception {
        mapper = new MapClass();
        context = mockery.mock(Mapper.Context.class);
        text = new Text("01-Apr-12 00:00:00,026  INFO  [scheduler_Worker-4] com.mycompany.sync.DistributedXipQuartzJobHandler.executeInternal(97) - Lock was acquired by other xip instance Quartz-lock [checkReversals] at 2012-04-01T00:00:00.026-07:00");
    }

    @Test
    public void testMap() throws Exception {
        mockery.checking(new Expectations() {
            {
                oneOf(context).write(with(new Text("scheduler_Worker-4")), with(new IntWritable(1)));
            }
        });

        mapper.map(new LongWritable(1), text, context);

        mockery.assertIsSatisfied();
    }
}

Hadoop Mapper.

2012-05-08T09:39:00.000-07:00

In this post, I make note of hadoop mapper, which are datastore containing key value pairs, similar to Map.

The basic thing to do while writing a mapper class is to extend Mapper. Note this class has type safety, hence pass in appropriate input and output key value pairs. Now override the method, map. The context object stores the output key value pair. Hence any logic extraction to be done is inside this map method.

An example: The mapper works over a log file which has following sample output:
01-Apr-12 00:00:00,026 INFO [scheduler_Worker-4] com.xoom.sync.DistributedXipQuartzJobHandler.executeInternal(97) - Lock was acquired by other xip instance Quartz-lock [bacCheckReversals] at 2012-04-01T00:00:00.026-07:00
The MapClass class's map method extracts anything between [], for the above the line it is scheduler_Worker-4 and stores the value in the context object as 1 (encapsulated in IntWritable object).

public static class MapClass extends Mapper<LongWritable, Text, Text, IntWritable> {
   public void map(LongWritable key, Text value, Context context) throws InterruptedException,IOException {
    String valueString = value.toString();
    String[] parts = valueString.split(" ");
    //fifth element in parts array should contain name of the thread being executed.
    // treat it as key.
    if (parts.length > 4 && parts[4] != null && parts[4].matches("^\\[.+\\]$")) {
     context.write(new Text(parts[4].replace("[","").replace("]","")),new IntWritable(1));
    } else {
     System.out.println("Value not collected by output collector:"+valueString);
    }
   }
  }

Next we move on to writing a simple Junit test case for the mapper class.

Passpharse less Ssh troubleshooting for Hadoop cluster setup.

2012-04-25T14:11:00.000-07:00

In this post, I want share my debugging knowledge gained while setting up passphrase less login. I needed this to setup a hadoop cluster. The instructions on the setting up hadoop are straight forward on how to generate one but it does not mention about troubleshooting. Hopefully this might serve one!

Recently I have been working on hadoop for analyzing production logs. As a proof of concept, I wanted my hadoop job to be run in production like environment (instead of pseudo distributed environment). Hence was setting up pass-phrase less ssh. Generated the public key on the name node and copied this key into another remote machine. Following are screenshots of the steps.

1) Generate the rsa private/public keys and copy to authorized_keys. This will help facilitate localhost passphrase less entry.

2) Copy public file to the remote server.

3) Append this key inside this file into authorized_keys. Create authorized_keys file if it does not exist.

After above steps, I was expected the setup should work and I tried to login. Surprisingly it failed, as it asked me to enter the password. After spending hour or so trying to figure what could be wrong, I came across this link which talks about how ssh client and server can be picky. In this page, the author mentions about file permissions. Hence I did quick check and it revealed that the permissions on local and remote files were different:

After changing the permissions, the login attempt succeeded.

If file permissions is not the issue in your case, here are few things that can be used to troubleshoot:
1) Check the public key is actually in authorized_keys remote node file.
2) Check for compatibility between ssh protocol version, solution is to force them to use version 2 using -2 flag.

Hibernate: Why cant you return Null...

2012-04-17T17:12:00.001-07:00

In this post, I rant about a patch the application needs to take care of when the framework is flawed. In this case, the framework happens to be Hibernate. The jar version used is : 3.6.6.Final
I was working on database query (removing sub-selects) optimizations and hence the need for refactoring the code. Below is the method which I came up with for the optimization.

protected long getMaxOrderingNumberForTrackingNo(String xoomTrackingNumber){
 List<Long> orderNumberList = getHibernateTemplate().find(HQL_GET_MAX_ORDER_NUMBER_FOR_TRACKING_NUMBER, xoomTrackingNumber);
 long orderNumber=0;
 if (!CollectionUtils.isEmpty(orderNumberList)) {
  orderNumber=orderNumberList.get(0);
 }
 return orderNumber;
}

After making my changes, I quickly ran the unit and integration tests. These tests to my surprise failed and I started the quest for finding the culprit! It turns out the issue was in the getHibernateTemplate().find() method which returned an instantiated object with size equals 1 and all elements in the list to be null. The above code hence threw a Null pointer exception. Below debugger screenshot from Idea confirms this claim.

Initially, I thought the CollectionUtils.isEmpty() method does not handle Collections.EMPTY_LIST.
Hence the method was changed to following:

protected long getMaxOrderingNumberForTrackingNo(String xoomTrackingNumber){
 List<Long> orderNumberList = getHibernateTemplate().find(HQL_GET_MAX_ORDER_NUMBER_FOR_TRACKING_NUMBER, xoomTrackingNumber);
 long orderNumber=0;
 if (orderNumberList!=null && !orderNumberList.isEmpty()) {
   orderNumber =  orderNumberList.get(0);
 }
 return orderNumber;
}

Still the test cases failed! Also my assumption about CollectionUtils.isEmpty method not handling Collections.EMPTY_LIST was wrong
Try out:

assertEquals(true, CollectionUtils.isEmpty(Collections.EMPTY_LIST));

So finally confirming that this is a flaw, did the following as a work-around, had to add in a try catch block and document this issue.

protected long getMaxOrderingNumberForTrackingNo(String xoomTrackingNumber){
 List<Long> orderNumberList = getHibernateTemplate().find(HQL_GET_MAX_ORDER_NUMBER_FOR_TRACKING_NUMBER, xoomTrackingNumber);
 long orderNumber=0;
 //Need to encapsulate below statement in a try-catch.
 //because stupid hibernate instantiates the List object with size=1 and its elements are empty.
 //This makes condition check !org.apache.commons.collections.CollectionUtils.isEmpty on the Collection useless.
 try {
   orderNumber=orderNumberList.get(0);
 } catch (NullPointerException npe) {
   orderNumber = 0;
   logger.debug("order number list is empty.");
 }
 return orderNumber;
}

In this case, it seems to make sense for the method to return in null so that commons utils can check for its emptiness. At this point, I will consider this as a hibernate framework bug.

Found where in hibernate, it is being initialized:
In SessionImpl.java, method list(String query, QueryParameters queryParameters), line 1263:

List results = CollectionHelper.EMPTY_LIST;

CollectionHelper's EMPTY_LIST constant is initialized as follows:

public static final List EMPTY_LIST = Collections.unmodifiableList( new ArrayList(0) );

From the screenshot above, note the size is 1 and List.isEmpty method checks on size == 0 which explains why these methods are failing! Below is the source code snippet of isEmpty method used in ArrayList.java

/**
 * Returns <tt>true</tt> if this list contains no elements.
 * @return <tt>true</tt> if this list contains no elements
 */
 public boolean isEmpty() {
  return size == 0;
 }

As expected, a quick unit test does also confirm that when a null element is added to List, the size is not 0. Following code fails on assertEquals.

List results = new ArrayList();
results.add(null);
assertEquals(0, results.size());

Deadlock, Starvation and Livelock.

2012-04-07T13:51:00.004-07:00

In this post, I will talk about an interesting deadlock and immediate steps taken to resolve it.

The system I work on is distributed in nature where instances communicate among themselves so they don't have to do the same work. We encountered an issue where batching was not being done.

On log analysis, it seemed the batch file was not generated because the thread couldn't acquire the necessary lock. This lock (a database lock) was held by another thread.

Being a self proclaimed concurrency expert in my team, I went straight into the machine where the webapp was running and executed jstack command to get the thread dump.

Following is the snippet of the thread dump which had acquired the lock:

"http-bio-41080-exec-10" daemon prio=10 tid=0x000000004d387000 nid=0x6620 runnable [0x0000000043e12000]
   java.lang.Thread.State: RUNNABLE
    at com.mycompany.integration.partner.CLASSNAME.createPartnerTransaction(CLASSNAME.java:72)
....

And following is the snippet of the thread dump which was waiting for the lock to be released:

"scheduler_Worker-8" prio=10 tid=0x00002aaab05c9000 nid=0x653d waiting for monitor entry [0x0000000044d25000]
   java.lang.Thread.State: BLOCKED (on object monitor)
    at com.mycompany.integration.service.impl.CLASSNAME1.acquireNonBlockingLock(CLASSNAME1.java:35)
    - waiting to lock <0x00000000c7bc4b60> (a java.lang.String)
....

Taking several thread dumps at regular intervals revealed that http-bio-41080-exec-10 thread was doing something fishy line 72 in CLASSNAME.createPartnerTransaction.
The line number 72 surrounding the block is as follows:

String filler = "";
while (accountNumber.length() + filler.length() < 9) {
     filler.concat(" ");
}
transaction.setPayeeAccountNumber(filler.concat(accountNumber));

Initial glance at the above block seemed fine, I asked myself what could be wrong with above while of code.... The issue was lurking which would only be executed when the account number was less than 9. Noticed it? On second look, the bug seemed obvious...

But just to reconfirm the issue, I took the jar file this class was present in and used a jad de-compiler to get back the java file.

The above block of code was converted into following line by compiler.

for(filler = ""; accountNumber.length() + filler.length() < 9; filler.concat(" "));

Noticed it??
Strings in java are immutable and result of concat method needs to be appended to the original variable (in this case, filler) to get the new value.

Immediate step to resolve this issue:
As it turns out, an infinite loop was present and would be executed if length of accountNumber+filler (always 0) would be less than 9. On detecting this issue, resolution was as simple as replacing the above block of code with apache commons StringUtils.padLeft.

Thoughts on improving this code on a long term basis:
The major issue here is not the infinite loop (thats a symptom), but I think its the combination of Starvation (couldn't acquire the lock held in database, which is the resource) and Livelock (retrying the operation of getting the resource which will always fail because it is acquired by another thread).

Was there a need to hold this lock when executing this method operation (or the nested methods for that matter)? This is one of the important question that we need to answer by revisiting its architecture....

Java's BigDecimal.

2011-11-02T22:47:00.000-07:00

Where to use Java's BigDecimal?
A example where it can be used is in applications which needs floating points with built in rounding facility.

Consider a case, where money is stored as a decimal in database. Sql mapping link states: "The recommended Java mapping for the DECIMAL and NUMERIC types is java.math.BigDecimal. The java.math.BigDecimal type provides math operations to allow BigDecimal types to be added, subtracted, multiplied, and divided with other BigDecimal types, with integer types, and with floating point types."

Lets say we want to compute a currency of another country. With currency conversion rate, a currency can be converted into required one. Using BigDecimal's divide method one can achieve it by mentioning the scale factor and rounding strategy.

2 is the scale (decides the number of digits after decimal point) and ROUND_HALF_EVEN strategy is banker's rounding.

BigDecimal amount = new BigDecimal(1);
BigDecimal rate = new BigDecimal(3);
BigDecimal destinationAmount = amount.divide(rate, 2, BigDecimal.ROUND_HALF_EVEN);

Mysql Stored Procedure for Computing Next Business day.

2011-10-24T16:44:00.000-07:00

In this post, I will explain how to compute next business day using stored procedure. To compute next business day, the code has to take into account Saturday, Sunday and any other bank holidays. Store the public holidays into bank_holiday table with column holiday_date of type DATETIME

Explanation: The first step is to get next day and then check if its a Saturday/Sunday, then increment accordingly (if Saturday increment by 2 days, if Sunday increment by 1 day). Next step involves checking this computed date exists in the bank_holiday table and then recursive calling itself for a match. Following is the code snippet.

DROP PROCEDURE IF EXISTS NextBusinessDay;

CREATE PROCEDURE NextBusinessDay (IN startDate DATETIME, OUT nextBDay DATETIME)

BEGIN
 DECLARE dayOfWeek INT DEFAULT 0;
 DECLARE rowCount INT DEFAULT 0;
 
 SELECT DATE_ADD(startDate, INTERVAL 1 DAY) INTO nextBDay;
 SELECT DAYOFWEEK(nextBDay) INTO dayOfWeek;
  
 IF dayOfWeek = 7 --Saturday then working day is Monday.
  THEN SELECT DATE_ADD(nextBDay, INTERVAL 2 DAY) INTO nextBDay;
 END IF;

 IF dayOfWeek = 1 --Sunday then working day is Monday.
  THEN SELECT DATE_ADD(nextBDay, INTERVAL 1 DAY) INTO nextBDay;
 END IF;

-- Check for holiday by querying the bank_holiday table.
 SELECT count(*) from bank_holiday where holiday_date = nextBDay INTO rowCount;

 IF rowCount > 0
  THEN CALL NextBusinessDay(nextBDay,nextBDay);
 END IF;
END

For detailed explaination on built-in DATE_ADD, DAYOFWEEK mysql functions, refer to link. For a short tutorial on writing stored procedure, refer to link.

Enabling recursive stored procedure: On executing this stored procedure, you will get an error. Following is the trace:

mysql> CALL NextBusinessDay('2010-07-02 00:00:00.0', @nextBDay);
ERROR 1456 (HY000): Recursive limit 0 (as set by the max_sp_recursion_depth variable) was exceeded for routine NextBusinessDay

Enable recursion by passing a int value greater than 0 on mysql server startup. Following is one way:

sudo -b /usr/local/mysql/bin/mysqld_safe --max_sp_recursion_depth=3

More at link.

Executing stored procedure: Using CALL. In this case, it will be CALL NextBusinessDay('2011-10-24', @nextBDay); To retrieve the result, do select @nextBDay; (i.e. Note this is the same variable passed into Stored procedure call) Following screenshot shows tests of this procedure over some dates. (Note in US: 5th Sept is observed as Memorial day and hence its a public holiday. Also tested for Leap year 2012.)

Proper selection of Hibernate Inheritance herirachy.

2011-10-17T14:06:00.001-07:00

In this post, I will explain a scenario which could have been avoided if proper hibernate inheritance strategy was selected in the first place. To begin with it was not even identified as Hibernate inheritance strategy issue. Infact IT noticed site slowness and reported about it to Engineering that a query was executing slowly (over 40 seconds). From the query which looked as below:

select column1,....,column50
from (
select .. from table1
union
select .. from table2
union
select .. from table3 ) where xid=1;

Can above query be optimized? Lets observer the query first. The above queries three different tables, then unions them and then filters the result using where clause. As you guessed it, querying individual table, union them and then filtering doesn't seem right. What happens if each table has 10K records, wont we be gathering 30K records before filtering on xid! Think what happens if there are millions of records. Besides that the query would be inefficient if queried on un-indexed column. This behavior was being observed in production where we have lots of records, queried on un-indexed column and hence cause for the query to be slow. This was confirmed on explaining the query. Below is the screen shot which confirms the behavior.

Ideally we would want to have a query (also indexing the column being queried) which is as follows:

select column1,....,column50
from (
select .. from table1 where  xid=1
union
select .. from table2 where  xid=1
union
select .. from table3 where  xid=1
);

This improves the query. Can be confirmed by explain on the query. From the below screenshot, we can confirm that lot less records are fetched before union. Also this time, column is indexed and hence the less time! :)

Again, this is was all Hibernate doing this stuff due to the fact that inheritance strategy selected was TABLE_PER_CLASS. As mentioned here, this strategy runs into join issue and hence the cause for being slow.

Solution adopted: Changed the strategy to SINGLE_TABLE (combined the data from three tables into one. The records are differentiated by a discriminator column). The only trade-off adopting SINGLE_TABLE strategy is that records can be sparse (i.e. not all columns will be populated and hence cannot enforce null constraint). Luckily that was not the case for us! :)

Configuring Mysql for remote tcp/ip connections.

2011-10-10T18:45:00.000-07:00

Scenario:
When a mysql server is running on a different machine than the required, following instructions can be of help. Not sure if these same instructions can be used for production. As most of my development is on mac, I have my mysql server running on it. There was recent feature configuration enhancement using IBM's CastIron tool. This tool apparently is only built for Windows environment. So I ended up having to install Virtual machine (VMware Fusion) with Windows Xp on it. For this feature enhancement, the tool needed to communicate to database (the very same db on which development is done on.) So there needed a way for this tool running on Windows needed to talk to database server running on Mac OSX.

Solution:
The solution is two folds. One create the user and second grant the user with appropriate permissions.
1) Create the user:
On mac, open the terminal and login into mysql. (mysql -u <user-name> -h <hostname> -p mysql). On successful mysql login, execute the command: create user '<username>'@'%' identified by 'password'; More info at link.

2) Grant the user:
Once above user is created, give the user grant (read, write or both). For example to give a less restrictive grant on this newly created user, execute the following command: grant all on *.* to '<username>'@'%'. More info on various parameters at link.

Test it using a mysql workbench on windows or the machine which you are trying to connect from.

Troubleshooting:
If for some reason, connection is not successful, check the following:

Check the my.cnf has line "skip-networking" commented.
Disable any firewall which might prevent the connection.

Apache Solr: Initial Encounter.

2011-08-29T23:11:00.000-07:00

In this post, I will try to explain how to setup tomcat with solr and working with import data from relational databases (oracle in my case). This guide is more of cookbook thing, more details can be found at solr-tomcat link and data import handler link. I am assuming tomcat is setup and running and hence wont go into details of setup but the link can be found here.

Configuring Solr.
1) Setup solr directory and setup variable $SOLR_HOME.
a) Download the apache solr source from link (click on Resources at left hand side tab and download the latest source code, apache-solr-3.3 at the time of writing this post).
b) Unzip/ untar the source code. Copy the following into separate directory (I copied it to /home/amrut/solr)

<SRC_DIR>/dist/apache-solr-3.3.war to /home/amrut/solr (cp ~/Downloads/apache-solr-3.3/dist/apache-solr.3.3.war /home/amrut/solr/solr.war)
Contents of <SRC_DIR>/example/solr/ to /home/amrut/solr (cp -r ~/Downloads/apache-solr-3.3/solr/ /home/amrut/solr/)

c) Setup environment variable $SOLR_HOME into .bashrc (for unix)/ .profile (for mac). export SOLR_HOME=/home/amrut/solr

2) Setup tomcat context fragment.
vi $CATALINA_HOME/conf/Catalina/localhost/solr.war

Paste the following into solr.war

<?xml version="1.0" encoding="utf-8"?>
 <Context docBase="/home/amrut/solr/solr.war" debug="0" crossContext="true">
 <Environment name="solr/home" type="java.lang.String" value="/home/amrut/solr/" override="true"/>
</Context>

3) Start tomcat and monitor the logs for any error.
Type the following at the command prompt: $CATALINA_HOME/bin/shutdown.sh; $CATALINA_HOME/bin/startup.sh; tail -f $CATALINA_HOME/logs/catalina.out

4) Solr should be available at: http://localhost:8080/solr.

Importing data from Relational database.
Consider a scenario where a table in mysql is to be imported into Solr. For people who want to try out, download the torrent dataset at link and insert into a table called torrent which has fields idtorrent, category, size, seeders, leechers. More information on how to load csv file into database can be found at link.
After loading the data into database table torrent, we must modify schema.xml and solrconfig.xml files.
1) schema.xml: Configure fields in Solr.


<field indexed="true" name="seeders" stored="true" type="string" />
<field indexed="true" name="lechers" stored="true" type="string" />
<field indexed="true" name="category" stored="true" type="text_general" />
<field indexed="true" name="id" required="true" stored="true" type="string" />

2) solrconfig.xml: Map data import handler and corresponding source it should look at (data-config.xml)

<requestHandler name="/dataimport" class="org.apache.solr.handler.dataimport.DataImportHandler">
    <lst name="defaults">
      <str name="config">/opt/solr/example/torrent-data-config.xml</str>
    </lst>
  </requestHandler>

3) torrent-data-config.xml: contains the source information (in our case, the torrent table information)


<datasource driver="com.mysql.jdbc.Driver" password="mysql" url="jdbc:mysql://localhost:3306/torrent_schema" user="mysql">
<document name="torrent">
<entity name="item" query="select idtorrent as id, category, size, seeders, lechers from torrent" />
<field column="ID" name="id" />
<field column="CATEGORY" name="category" />
<field column="SIZE" name="size" />
<field column="SEEDERS" name="seeders" />
<field column="LECHERS" name="lechers" />
</document>
</datasource>

Finally, import data using going to this link: http://localhost:8080/solr/dataimport?command=full-import . Monitor the catalina.out and also status can be checked at http://localhost:8080/solr/dataimport?command=status.

Junit: Testing Private Methods.

2011-08-08T14:46:00.001-07:00

I was researching on how to test private methods in junit and did a quick query. Few notable search results pointed me to a PrivateAccessor class and in particular invoke method. This class is available inside junit-addon jar available at the link.

So the basic idea is that this class bypasses Java modifier security/access specifier on methods and fields.
Following are the required input parameters:
1) Object which contains private methods.
2) Private method to be tested.
3) Parameter argument type array passed to that private method.
4) Values array passed to that private method.

The invoke method returns an Object, which I found that it contains the return value of the private method. Straightforward, isn't!

Following is the snippet of code, I used this class as follows:

package mycompany;

import org.junit.Test;
import static org.junit.Assert.assertTrue;
import junitx.util.PrivateAccessor;

import java.io.File;

public class MCExtractorTest {

 @Test
 public void testMerge() throws Exception {
  MCExtractor extractor = new MCExtractor("/prod/medlineMatch.xml", "/home/amrut/els/medline/cip");
  extractor.merge();
 }

 @Test
 public void testGetCollecIdFromDatabase() throws Throwable {
  MCExtractor extractor = new MCExtractor("/prod/medlineMatch.xml", "/home/amrut/els/medline/cip");
  Object object = PrivateAccessor.invoke(extractor, "getCollecIdFromDatabase", new Class[]{String.class}, new String[]{"S00090X"});
  assertTrue(object.toString().equals("JL-1s20S000071-1s20S00090X"));
 }
}

LRU Cache: Another way: By Extending LinkedHashMap

2011-07-31T23:18:00.000-07:00

In my previous post where I use PriorityQueue to construct a LRU cache. I was aware of LinkedHashMap data structure, but never provided great attention to its documentation. It turns out there is a much simpler way to construct an LRU cache using it.

The documentation clearly states,

"A special constructor is provided to create a linked hash map whose order of iteration is the order in which its entries were last accessed, from least-recently accessed to most-recently (access-order). This kind of map is well-suited to building LRU caches. Invoking the put or get method results in an access to the corresponding entry (assuming it exists after the invocation completes). The putAll method generates one entry access for each mapping in the specified map, in the order that key-value mappings are provided by the specified map's entry set iterator. No other methods generate entry accesses. In particular, operations on collection-views do not affect the order of iteration of the backing map

The removeEldestEntry(Map.Entry) method may be overridden to impose a policy for removing stale mappings automatically when new mappings are added to the map."

With this new discovery (at-least I became aware of the fact :) ), I wanted to try this out. Within few minutes, I was ready to test my code and as expected the code works. For purpose of testing, I have printed the eldest entry that will be removed once the size is reached. As you can see, I extended LinkedHashMap and overridden the method: removeEldestEntry.

Following is the code if someone wishes to try out.

package lru;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

public class Cache extends LinkedHashMap {
 
 private static int CACHE_SIZE=5;
 private static float LOAD_FACTOR = 0.75f;
 private static boolean ACCESS_ORDER = true;
 
 
 public Cache(int initialCapacity, float loadFactor, boolean accessOrder) {
  super(initialCapacity,loadFactor, accessOrder);
 }

 protected boolean removeEldestEntry(Map.Entry eldest) {
  if (size() > CACHE_SIZE) {
   System.out.println("Size limit reached, removing eldest entry:"+eldest.getKey());
   return true;
  } else {
   return false;
  }
 }
 
 public static void main (String[] args) {
  Cache cache = new Cache(CACHE_SIZE, LOAD_FACTOR , ACCESS_ORDER);
  cache.put("a","a");cache.put("b","b");
  cache.put("c","c");cache.put("d","d");
  cache.put("e","e");cache.put("b","b");
  cache.put("d","d");cache.put("b","b");
  cache.put("c","c");cache.put("b","b");
  cache.put("a","a");cache.put("f","f");
  cache.get("a");
  System.out.println("size:"+cache.size());
  Set keys = cache.keySet();
  for (String key : keys) {
   System.out.println("Cache contains:"+key);
  }
 }

}

Output produces is:

Size limit reached, removing eldest entry:e
size:5
Cache contains:d
Cache contains:c
Cache contains:b
Cache contains:f
Cache contains:a

Conclusion/Advantages:
1) Code is very much terse.
2) Reusing the code which is much stable, instead of re-inventing the world.

Nutch Custom Plugin: Initial Experience.

2011-07-31T22:18:00.000-07:00

I wanted to get familiar with plugin based model of Nutch and hence sharing my experience by writing custom plugin. In another post, I note my experience on debugging Nutch.

I followed tutorial which has been updated to 1.3 (latest at the time of writing this post). Plugins are to be created inside src/plugin directory.

Custom plugin needs following things:
1) Configuration files like build.xml, ivy.xml, plugin.xml.
2) Java classes: which execute business logic.

Following notes should be helpful in setting Nutch quickly.
1) Directory struture.
a) The main custom plugin directory should be defined inside <NUTCH_HOME_DIR>/src/plugin.
b) Inside this directory create build.xml and plugin.xml.
c) Now create directory src/java/<package-path>, in my case: src/java/com/mycompany/parserfilter/
d) In this directory, create a lib directory which store any external jars.

2) Make sure build.xml is present. It looks like this, replace <UR_PLUGIN_NAME> with plugin name.

<?xml version="1.0"?>
<project name="<UR_PLUGIN_NAME>" default="jar-core">
  <import file="../build-plugin.xml" />
</project>

3) Plugin.xml: This makes aware of custom code to Nutch system. Any other libraries which your custom code depends on can be defined inside libray tag. I wanted to extract certain section of the crawled page. Hence I am extending HtmlParseFilter. For every custom code, declaration is needed in the form of extension tag. I have defined TagExtractorParseFilter.

<?xml version="1.0" encoding="UTF-8"?>
<plugin id="food" name="<UR_PLUGIN_NAME>" version="0.0.1" provider-name="mycompany">
   <runtime>
      <library name="<UR_PLUGIN_NAME>">
         <export name="*"/>
      </library>
      <library name="jsoup.jar" />
   </runtime>
   <requires>
      <import plugin="nutch-extensionpoints" />
    </requires>
   <extension id="com.mycompany.parsefilter.TagExtractorParseFilter"
              name="<UR_PLUGIN_NAME>"
              point="org.apache.nutch.parse.HtmlParseFilter">
      <implementation id="TagExtractorParseFilter"
                      class="com.mycompany.parsefilter.TagExtractorParseFilter"/>
   </extension>
</plugin>

4) Remember to register this plugin in build.xml defined inside <NUTCH_HOME_DIR>/src/plugin/ . It can be registered : <ant dir="<UR_PLUGIN_NAME>" target="deploy" /> inside target tag. Failure to do so would result in your code not being packaged (into jar file). As of 1.3, the custom code is packaged into jar file and it resides in <NUTCH_HOME-DIR>/runtime/local/plugin/.

5) Lastly write your custom code:

package com.mycompany.parsefilter;

import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.parse.HTMLMetaTags;
import org.apache.nutch.parse.HtmlParseFilter;
import org.apache.nutch.parse.ParseResult;
import org.apache.nutch.protocol.Content;
import org.w3c.dom.*;


import org.apache.commons.logging.Log;
import org.jsoup.*;
import org.jsoup.select.Elements;

public class TagExtractorParseFilter implements HtmlParseFilter {

 private Configuration conf;
 public static final Log LOG = LogFactory.getLog(TagExtractorParseFilter.class);

 @Override
 public Configuration getConf() {
  return conf;
 }

 @Override
 public void setConf(Configuration arg0) {
  conf = arg0;
 }

 @Override
 public ParseResult filter(Content content, ParseResult parseResult,
   HTMLMetaTags metaTags, DocumentFragment doc) {
  org.jsoup.nodes.Document document =Jsoup.parse(content.toString());
  LOG.warn("document:"+document.text());
  Elements e = document.select("#rcpinglist");
  if (e != null)
   LOG.warn("ingredients:"+e.text());
  else
   LOG.warn("rcpinglist empty.");
  return parseResult;
 }
}

6) Running nutch which invokes custom code:
a) Create a seed list (the initial urls to fetch) :
mkdir urls

echo "http://www.YOURWEBSITE.com/" > urls/seed.txt
b) Inject seed url(s) to nutch crawldb (execute in <NUTCH_HOME_DIR>/runtime/local)
bin/nutch inject crawl/crawldb urls
c) Generate fetch list, fetch and parse content
bin/nutch generate crawl/crawldb crawl/segments
The above command will generate a new segment directory under crawl/segments that at this point contains files that store the url(s) to be fetched.
d) In the following commands we need the latest segment dir as parameter so we’ll store it in an environment variable:
export SEGMENT=crawl/segments/`ls -tr crawl/segments|tail -1`
e) Launch the fetcher that actually goes to get the content:
bin/nutch fetch $SEGMENT -noParsing
f) Next I parse the content:
bin/nutch parse $SEGMENT
g) Update the Nutch crawldb. The updatedb command wil store all new urls discovered during the fetch and parse of the previous segment into Nutch database so they can be fetched later. Nutch also stores information about the pages that were fetched so the same urls won’t be fetched again and again.
bin/nutch updatedb crawl/crawldb $SEGMENT -filter -normalize
h) Now a full Fetch cycle is completed. Next you can repeat steps (b-g) couple of times.

Notice in the logs, custom code will be invoked.

7) I would suggest reading my debug experience to get better understanding of the system, just in case you are stuck.

Debugging Nutch Parse Filter.

2011-07-23T23:19:00.000-07:00

Introduction:
I heard of Apache nutch, an open source crawler (well, actually it can do more than just crawling. It can index/search on the stored result, more info available at link). My curiosity to try this out led me through its code internals which I would like to share in this post. Particularly in this post I will share my debugging experience gained by stepping through its Parse segment internals.

Background:
I downloaded the latest stable code (at the time of this post, 1.3 being latest) from link. Custom plugin wiki is available at link. At the time of writing this post, the tutorial is yet to update for nutch 1.3. Also one can integrate nutch in eclispe using instructions at link, which is not yet complete.With little help available, I went the logging route, where I traced from the logs, what code was invoked. As of 1.3, Nutch can be run in two modes namely local and deploy. I tried the code in local, i.e. invoking nutch commands through command line inside dir <NUTCH_HOME_DIR>/runtime/local.

What was I trying to achieve?
I was trying to crawl/parse an Indian recipe website. From each of the recipe pages on this website, I wanted to extract Ingredients. Now one has two ways of going about this extraction, one which implements Parser or HtmlParseFilter. I choose HtmlParseFilter as Parser option is already taken care in HtmlParser plugin.

Approach:
From the wiki, I found I needed three things to get my code to work. One being the java class which actually did the parsing (TagExtractorParseFilter), plugin.xml and build.xml. More details on how to write a custom plugin can be found at link (will put the link as soon as I complete the post).
My first and main road block was my custom code was not invoked. I saw following in hadoop.log (which is located at NUTCH_HOME_DIR/runtime/local/logs/hadoop.log) Changing the log level to DEBUG to get more log information.

2011-07-23 00:33:37,592 WARN parse.ParserFactory - ParserFactory:Plugin: org.apache.nutch.parse.html.HtmlParser mapped to contentType application/xhtml+xml via parse-plugins.xml, but its plugin.xml file does not claim to support contentType: application/xhtml+xml
1962011-07-23 11:21:53,547 WARN parse.ParseUtil - Unable to successfully parse content http://www.tarladalal.com/ of type application/xhtml+xml

For the first comment, it turned out in my parse-plugins.xml had mapping of content type "application/xhtml+xml" but parse-html custom plugin's plugin.xml didnt state it. Had to make the following change, in order to make first error go away (in my case, it was an error as I need content type xhtml to be invoking HtmlParser code). First log is pretty intuitive and was resolved easily.

For Second log, I had to put log statements into code starting from ParseUtil -> HtmlParser -> HtmlParserFilters -> TagExtractorParseFilter (my custom code). Note my entry into ParseUtil was from the fact log statement in hadoop.log mentioned about it.

ParseUtil internally uses ParserFactory (based on Factory design pattern) to get the appropriate parser based on content type (note this content type is mapped in parse-plugins.xml file and it should match the content type in plugins plugin.xml). Once ParseUtil get hold of appropriate parser (in my case, it is HtmlParser), it calls

parseResult = p.getParse(content)

HtmlParser.getParse does a bunch of stuff before calling filters, invoked by

ParseResult filteredParse = this.htmlParseFilters.filter(content, parseResult, metaTags, root);

As these Filters, they are programmed to the interface, in my case, TagExtractorParseFilter.filter is invoked. On further investigation, it turned out there was a NPE thrown. Following is the code snippet where NPE was thrown.

@Override
public ParseResult filter(Content content, ParseResult parseResult, HTMLMetaTags metaTags, DocumentFragment doc) {
 Document document = doc.getOwnerDocument();
 NodeList nodeList = e.getChildNodes();
 Element e = document.getElementById("rcpinglist");
 int size = nodeList.getLength();
 for (int i=0; i<size; i++) {
  Node nodeItem = nodeList.item(i);
  LOG.warn("Node Value:"+nodeItem.getNodeValue());
 }
 return parseResult;
}

Note error thrown was a NPE, but the stack trace was never visible in either logs or stdout. As it turns out, the exception was being swallowed and parseResult.isSuccess() was set to false, indicating parsing had failed. Following code snippet from HtmlParseFilters, which shows how each filter is called and parseResult is obtained from each. Also on error during filter, original parseResult is returned.

/** Run all defined filters. */
public ParseResult filter(Content content, ParseResult parseResult, HTMLMetaTags metaTags, DocumentFragment doc) {
  // loop on each filter
  for (int i = 0 ; i < this.htmlParseFilters.length; i++) {
   // call filter interface
   parseResult = htmlParseFilters[i].filter(content, parseResult, metaTags, doc);
   // any failure on parse obj, return
   if (!parseResult.isSuccess()) {
   // TODO: What happens when parseResult.isEmpty() ?
   // Maybe clone parseResult and use parseResult as backup...
   // remove failed parse before return
   parseResult.filter();
   return parseResult;
  }
 }
 return parseResult;
}

Once code execution returned to ParseUtils, second log message was printed due to this error.Thus I modified my code to handle the NPE and that resolved my second log.

@Override
public ParseResult filter(Content content, ParseResult parseResult, HTMLMetaTags metaTags, DocumentFragment doc) {
 Document document = doc.getOwnerDocument();
 NodeList nodeList = e.getChildNodes();
 Element e = document.getElementById("rcpinglist");
 if (e != null) {
  int size = nodeList.getLength();
  for (int i=0; i<size; i++) {
   Node nodeItem = nodeList.item(i);
   LOG.warn("Node Value:"+nodeItem.getNodeValue());
  }
 } else {
  LOG.warn("no rcpinglist found.");
 }
 return parseResult;
}

Conclusion:
Log statements are very useful and give good insights on debugging the program. Using log as my starting point, I was able to understand the code in a much better way. The only con I feel is it takes little extra time. I would recommend this route if one needs to understand a system internals.

LRU Cache using PriorityQueue

2011-07-11T16:56:00.000-07:00

Background:
LRU (least recently used) cache is a good way of implementing cache because it frees up precious memory based on the last time an element was used. The criteria here is to purge items based on when the last time a particular element was last accessed.

In this post, I tried to implement a LRU cache using PriorityQueue. My rationale for choosing this combination: PriorityQueue does the sorting internally based on Comparator passed to it and I can concentrate on creating the cache, rather than worrying about maintaining the sort order.

Update: Check another way of implementing lru which is more terse found at link.

Implementation details:
I mainly have two Classes:
1) Element.java: This is the place holder for values in PriorityQueue (PQ) which has timestamp and value attributes. timestamp is to be updated when a element is accessed (put or get operations inside LRUCache)
2) LRUCache.java: This class uses PQ, elements are ordered in this object based on timestamp (the least timestamp would the first to be popped out). There is a cap SIZE of 5 on the cache. HashMap stores the actual value. LRUCache provides two public methods: get and put. Each of which is explained below:
a) get(String key) : Idea is as soon as the key is retrieved from the map, element key (this key matches to that of HashMap, key) in PriorityQueue should be updated with current timestamp.
b) put(String key, String value) : This method is responsible for checking if the key entry exist in map (in this case, it is a cache hit and hence update the PQ element with new timestamp). If the key does not exist in map, check if the map size. If greater than the cap set, remove the least recently used entry. If not greater, simply add this key to PQ.
c) There are a set of private methods (insert, remove and update) which operate on PQ. insert and remove are pretty intuitive, both methods use offer and poll methods provided by PQ.
d) Update operation is interesting. I knew that underlying data structure for PQ is heap. I was under the impression, that modifying elements (updating the timestamp) in PQ would initiate re-ordering of elements process. My assumption soon proved wrong when I saw the output not matching the required behavior. As it turned out, a heap would update only on inserting or removing operations and hence for my update operation, I had to delete and insert the entry(partially sorted tree) into PQ. Thus update operation is performed in O(n.logn) times (find a element takes linear time and then inserting that element takes logn time).

Following is the code snippet:

package lru;

import java.util.*;
import org.apache.commons.lang.StringUtils;

public class LRUCache {
private static final int SIZE = 5;
private Map<String, String> map = new HashMap<String, String>();

private PriorityQueue<Element> pq = new PriorityQueue<Element>(SIZE, new Comparator(){
 @Override
 public int compare(Object arg0, Object arg1) {
 //Ordering the elements as per timestamp.
 if (! (arg0 instanceof Element) || !(arg1 instanceof Element))
   return 0;
 Element e1 = (Element) arg0;
 Element e2 = (Element) arg1;
 return e1.getTimestamp().compareTo(e2.getTimestamp());
 }
});

private void insert(Element e) {
 System.out.println("Received Element: "+e.getValue());
 pq.offer(e);
}

private String remove() {
 Element leastUsed = pq.poll();
 if (leastUsed != null) {
  System.out.println("Removing least used element:"+leastUsed.getValue());
  System.out.println("This element was last used:"+leastUsed.getTimestamp());
  return leastUsed.getValue();
 }
 return "";
}

private void update(String mostRecentEleKey) {
 //update priority queue with most recent access.
 //Internal datastructure on PriorityQueue is Heap and it is partially sorted.
 //This means, any update on elements means to delete them and add them again.

 Iterator<Element> pqIterator = pq.iterator();
 while(pqIterator.hasNext()) {
 Element e = pqIterator.next();
  if(e.getValue().equals(mostRecentEleKey)) {
   pqIterator.remove();
   break;
  }
 }
 Element mostRecent = new Element();
 mostRecent.setTimestamp(new Date());
 mostRecent.setValue(mostRecentEleKey);
 insert(mostRecent);
}

public String get(String key) {
 String value = map.get(key);
 if (StringUtils.isNotEmpty(value)) {
  System.out.println("Updating "+key+" with current timestamp.");
  update(key);
 }
 return value;
}

public void put(String key, String value) {
 System.out.println("Before put opertaion, map size:"+map.size());
 if (map.containsKey(key)) {
  System.out.println("Cache hit on key:"+key+", nothing to insert!");
  update(key);
 } else {
  if(map.size() >= SIZE) {
   String leastUsedKey = remove();
   map.remove(leastUsedKey);
  }
  System.out.println("Element not present in Cache: "+key);
  Element e = new Element();
  e.setValue(key);
  e.setTimestamp(new Date());
  insert(e);
  map.put(key,value);
 }
 System.out.println("After put operation, following stats are generated:");
 System.out.println("Least used element:"+pq.peek().getValue()+", last used at:"+pq.peek().getTimestamp());
 System.out.println("map size:"+map.size());
}

public static void main(String[] args) {
 LRUCache cache = new LRUCache();
 cache.put("a","a");cache.put("b","b");
 cache.put("c","c");cache.put("d","d");
 cache.put("e","e");cache.put("b","b");
 cache.put("d","d");cache.put("b","b");
 cache.put("c","c");cache.put("b","b");
 cache.put("a","a");cache.put("f","f");
 cache.get("a");
}
}


package lru;

import java.util.Date;
import org.apache.commons.lang.builder.*;

public class Element {
private String value;
private Date timestamp;

public String getValue() {
 return value;
}

public void setValue(String value) {
 this.value = value;
}

public Date getTimestamp() {
 return timestamp;
}

public void setTimestamp(Date timestamp) {
 this.timestamp = timestamp;
}

public boolean equals(Element e) {
 return value.equals(e.getValue());
}

public int hashCode(){
 return new HashCodeBuilder(17,37).append(value).toHashCode();
}
}

This produces output:

Before put opertaion, map size:0
Element not present in Cache: a
Received Element: a
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:1
Before put opertaion, map size:1
Element not present in Cache: b
Received Element: b
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:2
Before put opertaion, map size:2
Element not present in Cache: c
Received Element: c
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:3
Before put opertaion, map size:3
Element not present in Cache: d
Received Element: d
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:4
Before put opertaion, map size:4
Element not present in Cache: e
Received Element: e
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Before put opertaion, map size:5
Cache hit on key:b, nothing to insert!
Received Element: b
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Before put opertaion, map size:5
Cache hit on key:d, nothing to insert!
Received Element: d
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Before put opertaion, map size:5
Cache hit on key:b, nothing to insert!
Received Element: b
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Before put opertaion, map size:5
Cache hit on key:c, nothing to insert!
Received Element: c
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Before put opertaion, map size:5
Cache hit on key:b, nothing to insert!
Received Element: b
After put operation, following stats are generated:
Least used element:a, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Before put opertaion, map size:5
Cache hit on key:a, nothing to insert!
Received Element: a
After put operation, following stats are generated:
Least used element:e, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Before put opertaion, map size:5
Removing least used element:e
This element was last used:Mon Jul 11 17:03:29 PDT 2011
Element not present in Cache: f
Received Element: f
After put operation, following stats are generated:
Least used element:d, last used at: Mon Jul 11 17:03:29 PDT 2011
map size:5
Updating a with current timestamp.
Received Element: a

Further work:
1) Need to make it generic.
2) Handling memory leak.