diff --git a/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java
index 86b6782..eb48846 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java
@@ -30,10 +30,12 @@
 
 /**
  * Base input adapter which reads data from persistence database through DATASTAX API and writes into output port(s).&nbsp;
- * Subclasses should provide implementation to get tuples and querying to retrieve data. 
+ * Subclasses should provide implementation to get tuples and querying to retrieve data.
  * <p>
  * This is an abstract class. Sub-classes need to implement {@link #queryToRetrieveData()} and {@link #getTuple(Row)}.
  * </p>
+ *
+ * @param <T>
  * @displayName Abstract Cassandra Input
  * @category Store
  * @tags input operator
@@ -73,17 +75,27 @@
   {
     String query = queryToRetrieveData();
     logger.debug(String.format("select statement: %s", query));
-
-    try {
-      ResultSet result = store.getSession().execute(query);
-      for(Row row: result) {
-        T tuple = getTuple(row);
-        outputPort.emit(tuple);
+    if (!"".equals(query)) {
+      try {
+        ResultSet result = store.getSession().execute(query);
+        processResult(result);
+      }
+      catch (Exception ex) {
+        store.disconnect();
+        DTThrowable.rethrow(ex);
       }
     }
-    catch (Exception ex) {
-      store.disconnect();
-      DTThrowable.rethrow(ex);
+  }
+
+  /*
+   * ProcessResult can be overridden in classes providing implementation of  AbstractCassandraInputOperator.
+   * Providing a basic implementation here which iterates through the rows in result,gets the particular tuple from row and emits it.
+   */
+  protected  void processResult(ResultSet result)
+  {
+    for (Row row: result) {
+          T tuple = getTuple(row);
+          outputPort.emit(tuple);
     }
   }
 }
diff --git a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
new file mode 100644
index 0000000..b1caa66
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
@@ -0,0 +1,442 @@
+/*
+ * Copyright (c) 2015 DataTorrent, Inc. ALL Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datatorrent.contrib.cassandra;
+
+import com.datastax.driver.core.ColumnDefinitions;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.util.PojoUtils;
+import com.datatorrent.lib.util.PojoUtils.Setter;
+import com.datatorrent.lib.util.PojoUtils.SetterBoolean;
+import com.datatorrent.lib.util.PojoUtils.SetterDouble;
+import com.datatorrent.lib.util.PojoUtils.SetterFloat;
+import com.datatorrent.lib.util.PojoUtils.SetterInt;
+import com.datatorrent.lib.util.PojoUtils.SetterLong;
+import java.math.BigDecimal;
+import java.util.*;
+import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * <p>
+ * CassandraInputOperator</p>
+ * A Generic implementation of AbstractCassandraInputOperator which gets field values from Cassandra database columns and sets in a POJO.
+ *
+ * @displayName Cassandra POJO Input Operator
+ * @category Input
+ * @tags input operator
+ */
+public class CassandraPOJOInputOperator extends AbstractCassandraInputOperator<Object>
+{
+  private List<String> columns;
+  private final transient List<DataType> columnDataTypes;
+  private Number startRow = 0;
+  @NotNull
+  private List<String> expressions;
+  @NotNull
+  private String tablename;
+  private final transient List<Object> setters;
+  @NotNull
+  private String retrieveQuery;
+  private transient Class<?> objectClass = null;
+  private boolean useAllColumns;
+  protected Number lastRowIdInBatch = 0;
+  @NotNull
+  protected String primaryKeyColumn;
+  protected transient DataType primaryKeyColumnType;
+
+  @Min(1)
+  private int limit = 10;
+
+  /*
+   * Number of records to be fetched in one time from cassandra table.
+   */
+  public int getLimit()
+  {
+    return limit;
+  }
+
+  public void setLimit(int limit)
+  {
+    this.limit = limit;
+  }
+
+  /*
+   * Primary Key Column of table.
+   * Gets the primary key column of Cassandra table.
+   */
+  public String getPrimaryKeyColumn()
+  {
+    return primaryKeyColumn;
+  }
+
+  public void setPrimaryKeyColumn(String primaryKeyColumn)
+  {
+    this.primaryKeyColumn = primaryKeyColumn;
+  }
+
+
+  /*
+   * User has the option to specify the starting row of the range of data they desire.
+   */
+  public Number getStartRow()
+  {
+    return startRow;
+  }
+
+  public void setStartRow(Number startRow)
+  {
+    this.startRow = startRow;
+  }
+
+  /*
+   * This option is for user to create POJO fields from a subset of columns in cassandra table.
+   */
+  public boolean isUseAllColumns()
+  {
+    return useAllColumns;
+  }
+
+  public void setUseAllColumns(boolean useAllColumns)
+  {
+    this.useAllColumns = useAllColumns;
+  }
+
+  /*
+   * POJO class which is generated as output from this operator.
+   * Example:
+   * public class TestPOJO{ int intfield; public int getInt(){} public void setInt(){} }
+   * outputClass = TestPOJO
+   * POJOs will be generated on fly in later implementation.
+   */
+  private String outputClass;
+
+  public String getOutputClass()
+  {
+    return outputClass;
+  }
+
+  public void setOutputClass(String outputClass)
+  {
+    this.outputClass = outputClass;
+  }
+
+  /*
+   * Query input by user: Example: select * from keyspace.tablename;
+   */
+  public String getRetrieveQuery()
+  {
+    return retrieveQuery;
+  }
+
+  public void setRetrieveQuery(String retrieveQuery)
+  {
+    this.retrieveQuery = retrieveQuery;
+  }
+
+  /*
+   * An ArrayList of Java expressions that will yield the cassandra column value to be set in output object.
+   * Each expression corresponds to one column in the Cassandra table.
+   */
+  public List<String> getExpressions()
+  {
+    return expressions;
+  }
+
+  public void setExpressions(List<String> expressions)
+  {
+    this.expressions = expressions;
+  }
+
+  /*
+   * Subset of columns specified by User in case POJO needs to contain fields specific to these columns only.
+   */
+  public List<String> getColumns()
+  {
+    return columns;
+  }
+
+  public void setColumns(List<String> columns)
+  {
+    this.columns = columns;
+  }
+
+  /*
+   * Tablename in cassandra.
+   */
+  public String getTablename()
+  {
+    return tablename;
+  }
+
+  public void setTablename(String tablename)
+  {
+    this.tablename = tablename;
+  }
+
+  public CassandraPOJOInputOperator()
+  {
+    super();
+    columnDataTypes = new ArrayList<DataType>();
+    setters = new ArrayList<Object>();
+    this.store = new CassandraStore();
+  }
+
+  @Override
+  public void setup(OperatorContext context)
+  {
+    super.setup(context);
+    if (setters.isEmpty()) {
+      try {
+        // This code will be replaced after integration of creating POJOs on the fly utility.
+        objectClass = Class.forName(outputClass);
+      }
+      catch (ClassNotFoundException ex) {
+        throw new RuntimeException(ex);
+      }
+
+      int numberOfColumns;
+
+      com.datastax.driver.core.ResultSet rs = store.getSession().execute("select * from " + store.keyspace + "." + tablename + " LIMIT " + 1);
+      ColumnDefinitions rsMetaData = rs.getColumnDefinitions();
+      if (!isUseAllColumns()) {
+        numberOfColumns = rsMetaData.size();
+      }
+      else {
+        numberOfColumns = columns.size();
+      }
+
+      primaryKeyColumnType = rsMetaData.getType(primaryKeyColumn);
+
+      for (int i = 0; i < numberOfColumns; i++) {
+        // Get the designated column's data type.
+        DataType type;
+        if (!isUseAllColumns()) {
+          type = rsMetaData.getType(i);
+        }
+        else {
+          type = rsMetaData.getType(columns.get(i));
+        }
+        columnDataTypes.add(type);
+        Object setter;
+        final String setterExpr = expressions.get(i);
+        switch (type.getName()) {
+          case ASCII:
+          case TEXT:
+          case VARCHAR:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, String.class);
+            break;
+          case BOOLEAN:
+            setter = PojoUtils.createSetterBoolean(objectClass, setterExpr);
+            break;
+          case INT:
+            setter = PojoUtils.createSetterInt(objectClass, setterExpr);
+            break;
+          case BIGINT:
+          case COUNTER:
+            setter = PojoUtils.createSetterLong(objectClass, setterExpr);
+            break;
+          case FLOAT:
+            setter = PojoUtils.createSetterFloat(objectClass, setterExpr);
+            break;
+          case DOUBLE:
+            setter = PojoUtils.createSetterDouble(objectClass, setterExpr);
+            break;
+          case DECIMAL:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, BigDecimal.class);
+            break;
+          case SET:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, Set.class);
+            break;
+          case MAP:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, Map.class);
+            break;
+          case LIST:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, List.class);
+            break;
+          case TIMESTAMP:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, Date.class);
+            break;
+          case UUID:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, UUID.class);
+            break;
+          default:
+            setter = PojoUtils.createSetter(objectClass, setterExpr, Object.class);
+            break;
+        }
+        setters.add(setter);
+      }
+    }
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public Object getTuple(Row row)
+  {
+    Object obj = null;
+    final int size = columnDataTypes.size();
+
+    try {
+      // This code will be replaced after integration of creating POJOs on the fly utility.
+      obj = objectClass.newInstance();
+    }
+    catch (InstantiationException ex) {
+      throw new RuntimeException(ex);
+    }
+    catch (IllegalAccessException ex) {
+      throw new RuntimeException(ex);
+    }
+
+    for (int i = 0; i < size; i++) {
+      DataType type = columnDataTypes.get(i);
+      switch (type.getName()) {
+        case UUID:
+          final UUID id = row.getUUID(i);
+          ((Setter<Object, UUID>)setters.get(i)).set(obj, id);
+          break;
+        case ASCII:
+        case VARCHAR:
+        case TEXT:
+          final String ascii = row.getString(i);
+          ((Setter<Object, String>)setters.get(i)).set(obj, ascii);
+          break;
+        case BOOLEAN:
+          final boolean bool = row.getBool(i);
+          ((SetterBoolean)setters.get(i)).set(obj, bool);
+          break;
+        case INT:
+          final int intValue = row.getInt(i);
+          ((SetterInt)setters.get(i)).set(obj, intValue);
+          break;
+
+        case BIGINT:
+        case COUNTER:
+          final long longValue = row.getLong(i);
+          ((SetterLong)setters.get(i)).set(obj, longValue);
+          break;
+        case FLOAT:
+          final float floatValue = row.getFloat(i);
+          ((SetterFloat)setters.get(i)).set(obj, floatValue);
+          break;
+        case DOUBLE:
+          final double doubleValue = row.getDouble(i);
+          ((SetterDouble)setters.get(i)).set(obj, doubleValue);
+          break;
+        case DECIMAL:
+          final BigDecimal decimal = row.getDecimal(i);
+          ((Setter<Object, BigDecimal>)setters.get(i)).set(obj, decimal);
+          break;
+        case SET:
+          Set<?> set = row.getSet(i, Object.class);
+          ((Setter<Object, Set<?>>)setters.get(i)).set(obj, set);
+          break;
+        case MAP:
+          final Map<?, ?> map = row.getMap(i, Object.class, Object.class);
+          ((Setter<Object, Map<?, ?>>)setters.get(i)).set(obj, map);
+          break;
+        case LIST:
+          final List<?> list = row.getList(i, Object.class);
+          ((Setter<Object, List<?>>)setters.get(i)).set(obj, list);
+          break;
+        case TIMESTAMP:
+          final Date date = row.getDate(i);
+          ((Setter<Object, Date>)setters.get(i)).set(obj, date);
+          break;
+        default:
+          throw new RuntimeException("unsupported data type " + type.getName());
+      }
+    }
+    return obj;
+  }
+
+  @Override
+  public String queryToRetrieveData()
+  {
+    boolean flag = false;
+
+    switch (primaryKeyColumnType.getName()) {
+      case INT:
+        if (startRow.intValue() > lastRowIdInBatch.intValue()) {
+          flag = true;
+        }
+        break;
+      case COUNTER:
+        if (startRow.longValue() > lastRowIdInBatch.longValue()) {
+          flag = true;
+        }
+        break;
+      case FLOAT:
+        if (startRow.floatValue() > lastRowIdInBatch.floatValue()) {
+          flag = true;
+        }
+        break;
+      case DOUBLE:
+        if (startRow.doubleValue() > lastRowIdInBatch.doubleValue()) {
+          flag = true;
+        }
+        break;
+    }
+
+    if (flag) {
+      return "";
+    }
+
+    startRow = lastRowIdInBatch.intValue() + 1;
+    StringBuilder sb = new StringBuilder();
+    sb.append(retrieveQuery).append(" where ").append("token(").append(primaryKeyColumn).append(")").append(">=").append(startRow).append(" LIMIT ").append(limit);
+    logger.debug("retrievequery is {}", sb.toString());
+
+    return sb.toString();
+  }
+
+  /*
+   * Overriding processResult to save primarykey column value from last row in batch.
+   */
+  @Override
+  protected void processResult(ResultSet result)
+  {
+      Row lastRowInBatch = null;
+        for (Row row: result) {
+          Object tuple = getTuple(row);
+          outputPort.emit(tuple);
+          lastRowInBatch = row;
+        }
+           if (lastRowInBatch != null) {
+          switch (primaryKeyColumnType.getName()) {
+            case INT:
+              lastRowIdInBatch = lastRowInBatch.getInt(0);
+              break;
+            case COUNTER:
+              lastRowIdInBatch = lastRowInBatch.getLong(0);
+              break;
+            case FLOAT:
+              lastRowIdInBatch = lastRowInBatch.getFloat(0);
+              break;
+            case DOUBLE:
+              lastRowIdInBatch = lastRowInBatch.getDouble(0);
+              break;
+            default:
+              throw new RuntimeException("unsupported data type " + primaryKeyColumnType.getName());
+          }
+        }
+  }
+
+  private static final Logger logger = LoggerFactory.getLogger(CassandraPOJOInputOperator.class);
+}
diff --git a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraOutputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOOutputOperator.java
similarity index 93%
rename from contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraOutputOperator.java
rename to contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOOutputOperator.java
index 3a1b048..1fe2e77 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraOutputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOOutputOperator.java
@@ -42,14 +42,15 @@
  * @tags output operator
  * @since 2.1.0
  */
-public class CassandraOutputOperator extends AbstractCassandraTransactionableOutputOperatorPS<Object>
+public class CassandraPOJOOutputOperator extends AbstractCassandraTransactionableOutputOperatorPS<Object>
 {
+  private static final long serialVersionUID = 201506181024L;
   @NotNull
   private ArrayList<String> columns;
-  private final ArrayList<DataType> columnDataTypes;
+  private final transient ArrayList<DataType> columnDataTypes;
   @NotNull
   private ArrayList<String> expressions;
-  private transient ArrayList<Object> getters;
+  private final transient ArrayList<Object> getters;
 
   /*
    * An ArrayList of Java expressions that will yield the field value from the POJO.
@@ -95,7 +96,7 @@
     this.tablename = tablename;
   }
 
-  public CassandraOutputOperator()
+  public CassandraPOJOOutputOperator()
   {
     super();
     columnDataTypes = new ArrayList<DataType>();
@@ -170,13 +171,13 @@
   {
     StringBuilder queryfields = new StringBuilder("");
     StringBuilder values = new StringBuilder("");
-    for (int i = 0; i < columns.size(); i++) {
+    for (String column: columns) {
       if (queryfields.length() == 0) {
-        queryfields.append(columns.get(i));
+        queryfields.append(column);
         values.append("?");
       }
       else {
-        queryfields.append(",").append(columns.get(i));
+        queryfields.append(",").append(column);
         values.append(",").append("?");
       }
     }
@@ -259,5 +260,5 @@
     return boundStmnt;
   }
 
-  private static transient final Logger LOG = LoggerFactory.getLogger(CassandraOutputOperator.class);
+  private static final Logger LOG = LoggerFactory.getLogger(CassandraPOJOOutputOperator.class);
 }
diff --git a/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java b/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
index 8a92788..a6d0d21 100644
--- a/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
+++ b/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
@@ -41,11 +41,13 @@
   public static final String KEYSPACE = "demo";
 
   private static final String TABLE_NAME = "test";
-  private static String APP_ID = "CassandraOperatorTest";
-  private static int OPERATOR_ID = 0;
+  private static final String TABLE_NAME_INPUT = "testinput";
+  private static final String APP_ID = "CassandraOperatorTest";
+  private static final int OPERATOR_ID = 0;
   private static Cluster cluster = null;
   private static Session session = null;
 
+  @SuppressWarnings("unused")
   private static class TestEvent
   {
     int id;
@@ -74,6 +76,8 @@
       session.execute(createMetaTable);
       String createTable = "CREATE TABLE IF NOT EXISTS " + KEYSPACE + "." + TABLE_NAME + " (id uuid PRIMARY KEY,age int,lastname text,test boolean,floatvalue float,doubleValue double,set1 set<int>,list1 list<int>,map1 map<text,int>,last_visited timestamp);";
       session.execute(createTable);
+      createTable = "CREATE TABLE IF NOT EXISTS " + KEYSPACE + "." + TABLE_NAME_INPUT + " (id int PRIMARY KEY,lastname text,age int);";
+      session.execute(createTable);
     }
     catch (Throwable e) {
       DTThrowable.rethrow(e);
@@ -86,6 +90,7 @@
     if (session != null) {
       session.execute("DROP TABLE " + CassandraTransactionalStore.DEFAULT_META_TABLE);
       session.execute("DROP TABLE " + KEYSPACE + "." + TABLE_NAME);
+      session.execute("DROP TABLE " + KEYSPACE + "." + TABLE_NAME_INPUT);
       session.close();
     }
     if (cluster != null) {
@@ -93,110 +98,91 @@
     }
   }
 
-
-  private static class TestOutputOperator extends CassandraOutputOperator
+  private static class TestOutputOperator extends CassandraPOJOOutputOperator
   {
+    private static final long serialVersionUID = 201506181038L;
+
     public long getNumOfEventsInStore()
     {
-        String countQuery = "SELECT count(*) from " + TABLE_NAME + ";";
-        ResultSet resultSetCount = session.execute(countQuery);
-        for (Row row: resultSetCount) {
-          return row.getLong(0);
-        }
-        return 0;
+      String countQuery = "SELECT count(*) from " + TABLE_NAME + ";";
+      ResultSet resultSetCount = session.execute(countQuery);
+      for (Row row: resultSetCount) {
+        return row.getLong(0);
+      }
+      return 0;
 
     }
 
     public void getEventsInStore()
     {
-        String recordsQuery = "SELECT * from " + TABLE_NAME + ";";
-        ResultSet resultSetRecords = session.execute(recordsQuery);
-        int count =0;
-        for (Row row: resultSetRecords) {
-          LOG.debug("Boolean value is {}", row.getBool("test"));
-          Assert.assertEquals(true, row.getBool("test"));
-          LOG.debug("lastname returned is {}", row.getString("lastname"));
-          Assert.assertEquals("abclast", row.getString("lastname"));
-          LOG.debug("Double value returned is {}", row.getDouble("doubleValue"));
-          Assert.assertEquals("Double value is",2.0,row.getDouble("doubleValue"),2);
-          LOG.debug("Float value returned is {}", row.getFloat("floatValue"));
-          LOG.debug("age returned is {}", row.getInt("age"));
-          LOG.debug("set returned is {} ", row.getSet("set1", Integer.class));
-          LOG.debug("list returned is {}", row.getList("list1", Integer.class));
-          LOG.debug("map returned is {}", row.getMap("map1", String.class, Integer.class));
-          LOG.debug("date returned is {}", row.getDate("last_visited"));
-          Assert.assertNotEquals(new Date(System.currentTimeMillis()),row.getDate("last_visited"));
-          if(count == 0)
-          {
-            Assert.assertEquals(2, row.getInt("age"));
-            Assert.assertEquals(2.0, row.getFloat("floatValue"),2);
-            Set<Integer> set = new HashSet<Integer>();
-            List<Integer> list = new ArrayList<Integer>();
-            Map<String,Integer> map = new HashMap<String, Integer>();
-            set.add(2);
-            list.add(2);
-            map.put("key2", 2);
-            Assert.assertEquals(set, row.getSet("set1", Integer.class));
-            Assert.assertEquals(map, row.getMap("map1", String.class, Integer.class));
-            Assert.assertEquals(list,row.getList("list1", Integer.class));
-          }
-          if(count == 1)
-          {
-            Assert.assertEquals(0, row.getInt("age"));
-            Assert.assertEquals(0.0, row.getFloat("floatValue"),2);
-            Set<Integer> set = new HashSet<Integer>();
-            List<Integer> list = new ArrayList<Integer>();
-            Map<String,Integer> map = new HashMap<String, Integer>();
-            set.add(0);
-            list.add(0);
-            map.put("key0", 0);
-            Assert.assertEquals(set, row.getSet("set1", Integer.class));
-            Assert.assertEquals(map, row.getMap("map1", String.class, Integer.class));
-            Assert.assertEquals(list,row.getList("list1", Integer.class));
-          }
-          if(count == 2)
-          {
-            Assert.assertEquals(1, row.getInt("age"));
-            Assert.assertEquals(1.0, row.getFloat("floatValue"),2);
-            Set<Integer> set = new HashSet<Integer>();
-            List<Integer> list = new ArrayList<Integer>();
-            Map<String,Integer> map = new HashMap<String, Integer>();
-            set.add(1);
-            list.add(1);
-            map.put("key1", 1);
-            Assert.assertEquals(set, row.getSet("set1", Integer.class));
-            Assert.assertEquals(map, row.getMap("map1", String.class, Integer.class));
-            Assert.assertEquals(list,row.getList("list1", Integer.class));
-          }
-          count++;
+      String recordsQuery = "SELECT * from " + TABLE_NAME + ";";
+      ResultSet resultSetRecords = session.execute(recordsQuery);
+      int count = 0;
+      for (Row row: resultSetRecords) {
+        LOG.debug("Boolean value is {}", row.getBool("test"));
+        Assert.assertEquals(true, row.getBool("test"));
+        LOG.debug("lastname returned is {}", row.getString("lastname"));
+        Assert.assertEquals("abclast", row.getString("lastname"));
+        LOG.debug("Double value returned is {}", row.getDouble("doubleValue"));
+        Assert.assertEquals("Double value is", 2.0, row.getDouble("doubleValue"), 2);
+        LOG.debug("Float value returned is {}", row.getFloat("floatValue"));
+        LOG.debug("age returned is {}", row.getInt("age"));
+        LOG.debug("set returned is {} ", row.getSet("set1", Integer.class));
+        LOG.debug("list returned is {}", row.getList("list1", Integer.class));
+        LOG.debug("map returned is {}", row.getMap("map1", String.class, Integer.class));
+        LOG.debug("date returned is {}", row.getDate("last_visited"));
+        Assert.assertNotEquals(new Date(System.currentTimeMillis()), row.getDate("last_visited"));
+        if (count == 0) {
+          Assert.assertEquals(2, row.getInt("age"));
+          Assert.assertEquals(2.0, row.getFloat("floatValue"), 2);
+          Set<Integer> set = new HashSet<Integer>();
+          List<Integer> list = new ArrayList<Integer>();
+          Map<String, Integer> map = new HashMap<String, Integer>();
+          set.add(2);
+          list.add(2);
+          map.put("key2", 2);
+          Assert.assertEquals(set, row.getSet("set1", Integer.class));
+          Assert.assertEquals(map, row.getMap("map1", String.class, Integer.class));
+          Assert.assertEquals(list, row.getList("list1", Integer.class));
         }
-
+        if (count == 1) {
+          Assert.assertEquals(0, row.getInt("age"));
+          Assert.assertEquals(0.0, row.getFloat("floatValue"), 2);
+          Set<Integer> set = new HashSet<Integer>();
+          List<Integer> list = new ArrayList<Integer>();
+          Map<String, Integer> map = new HashMap<String, Integer>();
+          set.add(0);
+          list.add(0);
+          map.put("key0", 0);
+          Assert.assertEquals(set, row.getSet("set1", Integer.class));
+          Assert.assertEquals(map, row.getMap("map1", String.class, Integer.class));
+          Assert.assertEquals(list, row.getList("list1", Integer.class));
+        }
+        if (count == 2) {
+          Assert.assertEquals(1, row.getInt("age"));
+          Assert.assertEquals(1.0, row.getFloat("floatValue"), 2);
+          Set<Integer> set = new HashSet<Integer>();
+          List<Integer> list = new ArrayList<Integer>();
+          Map<String, Integer> map = new HashMap<String, Integer>();
+          set.add(1);
+          list.add(1);
+          map.put("key1", 1);
+          Assert.assertEquals(set, row.getSet("set1", Integer.class));
+          Assert.assertEquals(map, row.getMap("map1", String.class, Integer.class));
+          Assert.assertEquals(list, row.getList("list1", Integer.class));
+        }
+        count++;
+      }
 
     }
 
   }
 
-  private static class TestInputOperator extends AbstractCassandraInputOperator<TestEvent>
+  private static class TestInputOperator extends CassandraPOJOInputOperator
   {
-
-    private static final String retrieveQuery = "SELECT * FROM " + KEYSPACE + "." + TABLE_NAME + ";";
-
-    @Override
-    public TestEvent getTuple(Row row)
-    {
-      try {
-        return new TestEvent(row.getInt(0));
-      }
-      catch (DriverException e) {
-        throw new RuntimeException(e);
-      }
-    }
-
-    @Override
-    public String queryToRetrieveData()
-    {
-      return retrieveQuery;
-    }
+    private final ArrayList<Integer> ids = new ArrayList<Integer>();
+    private final HashMap<Integer, String> mapNames = new HashMap<Integer, String>();
+    private final HashMap<Integer, Integer> mapAge = new HashMap<Integer, Integer>();
 
     public void insertEventsInTable(int numEvents)
     {
@@ -205,13 +191,14 @@
                 .addContactPoint(NODE).build();
         Session session = cluster.connect(KEYSPACE);
 
-        String insert = "INSERT INTO " + TABLE_NAME + " (ID)" + " VALUES (?);";
+        String insert = "INSERT INTO " + TABLE_NAME_INPUT + " (ID,lastname,age)" + " VALUES (?,?,?);";
         PreparedStatement stmt = session.prepare(insert);
         BoundStatement boundStatement = new BoundStatement(stmt);
-        Statement statement;
         for (int i = 0; i < numEvents; i++) {
-          statement = boundStatement.bind(i);
-          session.execute(statement);
+          ids.add(i);
+          mapNames.put(i, "test" + i);
+          mapAge.put(i, i + 10);
+          session.execute(boundStatement.bind(i, "test" + i, i + 10));
         }
       }
       catch (DriverException e) {
@@ -219,6 +206,21 @@
       }
     }
 
+    public ArrayList<Integer> getIds()
+    {
+      return ids;
+    }
+
+    public HashMap<Integer, String> getNames()
+    {
+      return mapNames;
+    }
+
+    public HashMap<Integer, Integer> getAge()
+    {
+      return mapAge;
+    }
+
   }
 
   @Test
@@ -284,10 +286,13 @@
     outputOperator.getEventsInStore();
   }
 
-  //This Test needs to be improved.
+  /*
+   * This test can be run on cassandra server installed on node17.
+   */
   @Test
   public void TestCassandraInputOperator()
   {
+    String retrieveQuery = "SELECT * FROM " + KEYSPACE + "." + TABLE_NAME_INPUT;
     CassandraStore store = new CassandraStore();
     store.setNode(NODE);
     store.setKeyspace(KEYSPACE);
@@ -298,22 +303,49 @@
 
     TestInputOperator inputOperator = new TestInputOperator();
     inputOperator.setStore(store);
-    inputOperator.insertEventsInTable(10);
+    inputOperator.setOutputClass("com.datatorrent.contrib.cassandra.TestInputPojo");
+    inputOperator.setTablename(TABLE_NAME_INPUT);
+    inputOperator.setRetrieveQuery(retrieveQuery);
+    ArrayList<String> columns = new ArrayList<String>();
+    columns.add("id");
+    columns.add("age");
+    columns.add("lastname");
 
+    inputOperator.setColumns(columns);
+    ArrayList<String> expressions = new ArrayList<String>();
+    expressions.add("id");
+    expressions.add("age");
+    expressions.add("lastname");
+    inputOperator.setExpressions(expressions);
+
+    inputOperator.insertEventsInTable(30);
+    inputOperator.setPrimaryKeyColumn("id");
     CollectorTestSink<Object> sink = new CollectorTestSink<Object>();
     inputOperator.outputPort.setSink(sink);
-
     inputOperator.setup(context);
     inputOperator.beginWindow(0);
+    inputOperator.insertEventsInTable(10);
     inputOperator.emitTuples();
     inputOperator.endWindow();
-
     Assert.assertEquals("rows from db", 10, sink.collectedTuples.size());
+    inputOperator.beginWindow(1);
+    inputOperator.emitTuples();
+    inputOperator.endWindow();
+    Assert.assertEquals("rows from db", 20, sink.collectedTuples.size());
+    ArrayList<Integer> listOfIDs = inputOperator.getIds();
+    // Rows are not stored in the same order in cassandra table in which they are inserted.
+    for (int i = 0; i < 10; i++) {
+      TestInputPojo object = (TestInputPojo)sink.collectedTuples.get(i);
+      Assert.assertTrue("id set in testpojo", listOfIDs.contains(object.getId()));
+      Assert.assertEquals("name set in testpojo", inputOperator.getNames().get(object.getId()), object.getLastname());
+      Assert.assertEquals("age set in testpojo", inputOperator.getAge().get(object.getId()).intValue(), object.getAge());
+    }
+
   }
 
   public static class TestPojo
   {
-    private TestPojo(UUID randomUUID, int i, String string, boolean b, float d, double d0, Set<Integer> set1, List<Integer> list1, Map<String, Integer> map1, Date date)
+    public TestPojo(UUID randomUUID, int i, String string, boolean b, float d, double d0, Set<Integer> set1, List<Integer> list1, Map<String, Integer> map1, Date date)
     {
       this.id = randomUUID;
       this.age = i;
@@ -440,6 +472,6 @@
 
   }
 
-  private static transient final Logger LOG = LoggerFactory.getLogger(CassandraOperatorTest.class);
+  private static final Logger LOG = LoggerFactory.getLogger(CassandraOperatorTest.class);
 
 }
diff --git a/contrib/src/test/java/com/datatorrent/contrib/cassandra/TestInputPojo.java b/contrib/src/test/java/com/datatorrent/contrib/cassandra/TestInputPojo.java
new file mode 100644
index 0000000..3a20406
--- /dev/null
+++ b/contrib/src/test/java/com/datatorrent/contrib/cassandra/TestInputPojo.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2015 DataTorrent, Inc. ALL Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datatorrent.contrib.cassandra;
+
+
+/*
+ * TestInputPojo class specified by user for specifying schema as part of testing.
+ * CassandraPOJOInputOperator can be used to get Field values from cassandra database columns
+ * and set in this pojo as an example.
+ */
+public class TestInputPojo
+{
+  private int id;
+
+  public int getId()
+  {
+    return id;
+  }
+
+  public void setId(int id)
+  {
+    this.id = id;
+  }
+  private String lastname;
+  private int age;
+
+
+  public int getAge()
+  {
+    return age;
+  }
+
+  public void setAge(int age)
+  {
+    this.age = age;
+  }
+
+  public String getLastname()
+  {
+    return lastname;
+  }
+
+  public void setLastname(String lastname)
+  {
+    this.lastname = lastname;
+  }
+
+}