My Cassandra Templates

So I thought I would share some of my abstract adapter code for, eh, interacting with Hector client, that in turn interacts with Cassandra.

The idea is it adapts any model/entity POJO class into a key and columns that fit a column family that corresponds to the model. This happens through the marshal and unmarshal methods that are specific to each implementor of this trait. Curious names--marshal and unmarshal, but I just don't want to call them map and unmap or pack and unpack because in Scala they have different concepts. Marshal transforms the model into, really, a Scala tuple of key and list of tuples of column name and value. The persist methods in this adapter know how to work with them. Unmarshal transforms query results (in the form of ColumnFamilyResult) into the model. And this one is used by query methods.

As you can see the adapter also functions as a high-level DAO in that the user of the DAO shouldn't know about how to work with Hector/Cassandra.

Fine, I know it gets roundabout, the interaction between this trait and its subclass, but it's just how it is for now. I don't want to spend too much time refining for now.

Furthermore there is this automatic Id field finder method (getKeyFieldOpt). It does some reflection work against the model to find which field is the Id (row key). In my case Id is a case class (like var id = Id(9999)). And I think I use some implicit defs so retrieving id transforms it automatically to its value whenever relevant. But you get the gist.

This is a work in progress. At the moment all generic queries work to fetch the entire row of a column family and unmarshal its data to a model, but I'll be working on generic slices and ranges soon.

So here's the base trait:

import java.lang.reflect.Field
import scala.collection.JavaConversions._
import id.pronimo.watchlet.exception.NoExistingRowException
import id.pronimo.watchlet.exception.WrongModelException
import javax.validation.ConstraintViolation
import javax.validation.ConstraintViolationException
import javax.validation.Validation
import javax.validation.ValidationException
import me.prettyprint.cassandra.serializers.AsciiSerializer
import me.prettyprint.cassandra.serializers.StringSerializer
import me.prettyprint.cassandra.serializers.UUIDSerializer
import me.prettyprint.cassandra.serializers.CompositeSerializer
import me.prettyprint.cassandra.serializers.DateSerializer
import me.prettyprint.cassandra.service.template.ColumnFamilyResult
import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate
import me.prettyprint.cassandra.service.template.ColumnFamilyUpdater
import me.prettyprint.hector.api.exceptions.HectorException
import me.prettyprint.hector.api.factory.HFactory
import me.prettyprint.hector.api.mutation.Mutator
import me.prettyprint.hector.api.Keyspace
import id.pronimo.watchlet.model.metadata.Id
import me.prettyprint.cassandra.serializers.SerializerTypeInferer
import me.prettyprint.hector.api.Serializer

trait CommonStandardCFAdapter[R <: Serializable, K, N] {
 @transient val (us, ss, as, ds, cs) = (UUIDSerializer.get, StringSerializer.get, AsciiSerializer.get, DateSerializer.get, CompositeSerializer.get)
 @transient val EMPTY_BYTE_ARRAY = Array.empty[Byte]
 @transient val validator = Validation.buildDefaultValidatorFactory.getValidator
 
 def fs[F](x:F) = { SerializerTypeInferer.getSerializer[F](x) }
 val hcolumn = (n:N, v:Any) => {
  HFactory.createColumn(n, v, getKeyspace.createClock, getNameSerializer, fs(v))
 }

 /* CONCRETE METHODS [BEGIN] */
 /**
  * Add a new record, or override existing data.
  *
  *  @tparam   R the record type defined per implementation
  */
 @throws(classOf[HectorException])
 @throws(classOf[ValidationException])
 def write(record: R, deleteFirst: Boolean) {
  validate(record)
  if (deleteFirst) remove(record)
  val thriftTemplate = this.getThriftTemplate
  val (k, c) = marshal(record)
  val updater = thriftTemplate.createUpdater(k)
  c.iterator.foreach {
   case (n, v) =>
    updater.setColumn(hcolumn(n, v))
  }
  thriftTemplate.update(updater)
 }

 /**
  * Update a record by specifying intended column to add/update.
  *
  * Will throw an exception when record with specified key does not exist.
  *
  *  @note  Use with caution, this method allows adding a column not defined in the record type.
  *  @param    key of the record
  *  @param    colName the column name
  *  @param    value the column value
  *  @tparam   K the key type defined per implementation
  *  @tparam   V the value type
  */
 @throws(classOf[HectorException])
 @throws(classOf[ValidationException])
 def update[V](key: K, colName: N, value: V) {
  val thriftTemplate = this.getThriftTemplate

  if (false == thriftTemplate.isColumnsExist(key)) {
   throw new NoExistingRowException(key, "update")
  }
  val updater = thriftTemplate.createUpdater(key)

  val column = HFactory.createColumn(colName, value, getKeyspace.createClock)
  updater.setColumn(column)
  thriftTemplate.update(updater)
 }

 /**
  * Update a record by specifying multiple intended columns to add/update.
  *
  * Will throw an exception when record with specified key does not exist.
  *
  *  @note  Use with caution, this method allows adding columns not defined in the record type/column family metadata.
  *  @param    key of the record
  *  @param    map of column name/column value
  *  @tparam   K the key type defined per implementation
  */
 @throws(classOf[HectorException])
 @throws(classOf[ValidationException])
 def update(key: K, map: Map[N, Any]) {
  val thriftTemplate = this.getThriftTemplate

  if (false == thriftTemplate.isColumnsExist(key)) {
   throw new NoExistingRowException(key, "update")
  }

  val updater = {
   thriftTemplate.createUpdater(key)
  }
  map.iterator.foreach {
   case (n, v) =>
    val column = HFactory.createColumn(n, v, getKeyspace.createClock)
    updater.setColumn(column)
  }
  thriftTemplate.update(updater)
 }

 /**
  * Add multiple records, or override when such with the same key exists.
  *
  * Will throw an exception when record with specified key does not exist, automatically cancelling whole operation.
  *
  *  @note  Use with caution, this method allows adding columns not defined in the record type/column family metadata.
  *  @param    key of the record
  *  @param    map of column name/column value
  *  @tparam   K the key type defined per implementation
  */
 @throws(classOf[HectorException])
 @throws(classOf[ValidationException])
 def writeBatch(records: List[R], deleteFirst: Boolean) {
  val cf = this.getThriftTemplate.getColumnFamily
  val thriftTemplate = this.getThriftTemplate

  records.iterator.foreach { it => validate(it) }

  val mutator = thriftTemplate.createMutator

  records.iterator.foreach { it =>
   if (deleteFirst) remove(it)
   marshal(it) match {
    case (k, c) => c.iterator.foreach { tuple =>
     mutator.addInsertion(k, cf, HFactory.createColumn(tuple._1, tuple._2, getKeyspace.createClock))
    }
   }
  }
  thriftTemplate.executeBatch(mutator)
 }

 /**
  * Read record by key, returning Option
  *
  *  @param    key of the record
  *  @tparam   K the key type defined per implementation
  *  @tparam   R the record type defined per implementation
  */
 @throws(classOf[HectorException])
 def read(key: K): Option[R] = {
  val thriftTemplate = this.getThriftTemplate
  val result = thriftTemplate.queryColumns(key)
  unmarshal(result)
 }

 /**
  * Remove entire row by key
  *
  *  @param    key of the record
  *  @tparam   K the key type defined per implementation
  */
 @throws(classOf[HectorException])
 def remove(key: K) {
  this.getThriftTemplate().deleteRow(key)
 }

 /**
  * Remove entire row using the record's id
  *
  *  @tparam   R the record type defined per impl
  *  ementation
  */
 @throws(classOf[HectorException])
 def remove(record: R) {
  remove(getKey(record))
 }

 /**
  * Get key field option (thru reflection)
  */
 protected def getKeyFieldOpt[E <: R: Manifest] = implicitly[Manifest[E]].erasure.getDeclaredFields.find(_.getType.equals(classOf[Id[K]]))
 
 /**
  * Get N serializer (thru reflection)
  */
 protected def getNSerializer[E <: N: Manifest]():Serializer[N] = SerializerTypeInferer.getSerializer[N](implicitly[Manifest[E]].erasure)

 /**
  * Get key for record (thru reflection)
  */
 def getKey[E <: R: Manifest](record: R): K = {
  try { getKeyField.get(record).asInstanceOf[Id[K]]}
  catch {
   case e =>
    throw new WrongModelException(implicitly[Manifest[E]].erasure, "Key type specified in template does not match actual key type of record.", e)
  }
 }

 /**
  * Validate record using JSR-303
  *
  *  @param    record
  *  @tparam   R the record type defined per implementation
  */
 @throws(classOf[ValidationException])
 def validate(record: R) {
  val violations = validator.validate(record)

  if (false == violations.isEmpty()) {
   val exceptionMessage = violations.iterator
    .map(it => it.getPropertyPath.toString + " " + it.getMessage)
    .mkString(", ")

   throw new ConstraintViolationException(exceptionMessage, violations.asInstanceOf[java.util.Set[ConstraintViolation[_]]])
  }
 }
 /* CONCRETE METHODS [END] */

 /* ABSTRACT METHODS [BEGIN] */
 /**
  * Transform record into tuples with format (key, List[(columnName, columnValue)])
  */
 def marshal(record: R): (K, List[(N, Any)])

 /**
  * Transform ColumnFamilyResult into record
  */
 def unmarshal(columns: ColumnFamilyResult[K, N]): Option[R]

 /**
  * Get underlying Thrift template used by this spec template
  */
 def getThriftTemplate(): ColumnFamilyTemplate[K, N]
 
 /**
  * Get column name serializer used by this spec template
  * This became mandatory for creating composite columns using the Hector client.
  */
 def getNameSerializer():Serializer[N]

 def getKeyField: Field

 def getKeyspace: Keyspace
 /* ABSTRACT METHODS [END] */
}

And a sample implementation:
PS: Don't mind the CDI annotations. And the transient annotations are there just so it plays along well with CDI.

@Named( "ItemRepository" )
@ApplicationScoped
class ItemCFAdapter extends CommonStandardCFAdapter[Item, UUID, String] with Serializable {
 @transient val keyField = {
  getKeyFieldOpt[Item] match {
   case Some( field ) =>
    field.setAccessible( true )
    field
   case None =>
    throw new WrongModelException( classOf[Item], "Cannot find field with Id metadata." )
  }
 }

 @transient val nameSerializer = getNSerializer[String]

 @Inject
 @transient
 var keyspace: Keyspace = _

 var thriftTemplate: ThriftColumnFamilyTemplate[UUID, String] = _

 @PostConstruct
 def initialise() {
  thriftTemplate = new ThriftColumnFamilyTemplate( keyspace, CF.CF_NAME, us, as )
 }

 override def marshal( record: Item ): ( UUID, List[( String, Any )] ) = {
  ( getKey( record ), List(
   ( "label", record.label ),
   ( "url", record.url ) ) )
 }

 override def unmarshal( columns: ColumnFamilyResult[UUID, String] ): Option[Item] = {
  val l = columns.getString( "label" )
  val u = columns.getString( "url" )
  ( Option( l ), Option( u ) ) match {
   case ( Some( _ ), Some( _ ) ) =>
    Option( new Item {
     id = columns.getKey
     label = l
     url = u    } )
   case _ =>
    None
  }
 }

 override def getThriftTemplate() = thriftTemplate

 override def getKeyField() = keyField

 override def getNameSerializer() = nameSerializer

 override def getKeyspace() = keyspace

}

My Little Take on Backend Development

Update 29-10-2012:
I found this blog post more comprehensively in defense of Java EE 6.
Original post:
This comes from someone who only just recently read about JTA and JTS. Take this with a grain of salt, so to speak.

I have some experience using the Spring Framework, mainly the Spring MVC sometime about 2years ago, and some again just recently (as indicated in my previous post).  Back then, it was first a multifinancing application, second a message broadcasting engine, sort of.  Neither of which I created from the ground up.  I was merely adding features and so on.  I couldn't compare it against the Java EE 6 at that time, lack of understanding of their internals and so on.

The message broadcasting engine had to be improved so that it could support more load.  There were topics such as clustering, high availability, parallelism, concurrency, etc.  I think it was then that I came across the JMS and some asynchronous processing features provided by the EJB 3.  That was to be my gateway to Java EE 6.   Afterwards it was a haphazard series of choice to learn the actor model, CSP, Scala, Erlang, Haskell, NoSQL, etc.

So I wonder what was it again, Spring Framework's winning points.  Their documentation is among the best.  Their framework itself is rock solid, for what it is.  What it is, really, is a little confusing, though.  It appears to be sort of an alternative to the Java EE stack, but heavily relies on it.  It manages the object lifecycles.  Java EE does that as well, manage object lifecycles.  Two things manage your object lifecycles.  Obviously you can run Spring Framework on a standard servlet container, but I see everywhere that it is recommended that you use, say, the JTA provided by the container when available.  It would be unavoidable that both containers will manage the object lifecycles if you used CMT alongside Spring.  And how is it that one could share a bean across deployments in Spring?  Even if you could, I think using Spring remoting, your object lifecycles management would be more complicated because it would be a communication between two unrelated containers.  It looks to me that the EJB specification is superior in this regard, with its local and remote beans concept and tighter integration with JTA and the rest of the Java EE specification.

Anyway, honestly I'm not a big fan of dependency injection, Spring or CDI.  It's a little too much magic for me.  Language level singleton object construction available in Scala and programmatic lookup and vending are easier to work with.

Around the Web

Sometime ago I checked out different approaches to web application development (in my free time.)  With Python I explored the Pyramid framework as well as Django.  With Scala I did Lift.  And lastly I explored JSF using Groovy as a replacement for Java.

Actually I was quite overwhelmed with Python.  It's not difficult to write anything in Python. 
However, to write good code in Python it should take considerably more effort and more thorough understanding of the language.  I stumbled across a lot of metaprogramming then.  I think it grossly difficult to feel secure because, with metaprogramming alongside dynamic typing, most of your application's survival depends on the condition of your gestalt.

Scala lifts much of that mental burden off me I should think.  I like Lift.  I could do more in Lift in the future.  :)

Groovy is, as I probably wrote in my previous post, somewhat in the middle when it comes to typing--a degree of static as well as dynamic typing.  It saves a lot of time and code lines.  In a way it intuitively guides me to write better code.  There can be a lot of metaprogramming too in Groovy, but it's not a show stopper.  You don't necessarily have to understand it to write good code.  Though it's pretty easy (I think compared to that in Python).  It would be much better if Java EE and relevant stuff were tailored to fit Groovy more.  And better IDE support.  And a way to filter exception stack traces (Seam Catch maybe?).