diff --git a/src/main/java/com/jmethods/catatumbo/DatastoreAccess.java b/src/main/java/com/jmethods/catatumbo/DatastoreAccess.java
index 7532bbe..f46cea2 100644
--- a/src/main/java/com/jmethods/catatumbo/DatastoreAccess.java
+++ b/src/main/java/com/jmethods/catatumbo/DatastoreAccess.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo;
 
+import java.lang.reflect.Type;
 import java.util.List;
 
 /**
@@ -37,6 +38,20 @@ public interface DatastoreAccess {
    */
   <E> E insert(E entity);
 
+  /**
+   * Inserts the given entity into the Cloud Datastore.
+   *
+   * @param entity
+   *          the entity to insert
+   * @param entityType
+   *          the entity type
+   * @return the inserted entity. The inserted entity will not be same as the passed in entity. For
+   *         example, the inserted entity may contain any generated ID, key, parent key, etc.
+   * @throws EntityManagerException
+   *           if any error occurs while inserting.
+   */
+  <E> E insert(E entity, Type entityType);
+
   /**
    * Inserts the given list of entities into the Cloud Datastore.
    * 
@@ -50,6 +65,21 @@ public interface DatastoreAccess {
    */
   <E> List<E> insert(List<E> entities);
 
+  /**
+   * Inserts the given list of entities into the Cloud Datastore.
+   *
+   * @param entities
+   *          the entities to insert.
+   * @param entityType
+   *          the entity type
+   * @return the inserted entities. The inserted entities will not be same as the passed in
+   *         entities. For example, the inserted entities may contain generated ID, key, parent key,
+   *         etc.
+   * @throws EntityManagerException
+   *           if any error occurs while inserting.
+   */
+  <E> List<E> insert(List<E> entities, Type entityType);
+
   /**
    * Updates the given entity in the Cloud Datastore. The passed in Entity must have its ID set for
    * the update to work.
@@ -62,6 +92,20 @@ public interface DatastoreAccess {
    */
   <E> E update(E entity);
 
+  /**
+   * Updates the given entity in the Cloud Datastore. The passed in Entity must have its ID set for
+   * the update to work.
+   *
+   * @param entity
+   *          the entity to update
+   * @param entityType
+   *          the entity type
+   * @return the updated entity.
+   * @throws EntityManagerException
+   *           if any error occurs while updating.
+   */
+  <E> E update(E entity, Type entityType);
+
   /**
    * Updates the given list of entities in the Cloud Datastore.
    * 
@@ -74,6 +118,20 @@ public interface DatastoreAccess {
    */
   <E> List<E> update(List<E> entities);
 
+  /**
+   * Updates the given list of entities in the Cloud Datastore.
+   *
+   * @param entities
+   *          the entities to update. The passed in entities must have their ID set for the update
+   *          to work.
+   * @param entityType
+   *          the entity type
+   * @return the updated entities
+   * @throws EntityManagerException
+   *           if any error occurs while inserting.
+   */
+  <E> List<E> update(List<E> entities, Type entityType);
+
   /**
    * Updates or inserts the given entity in the Cloud Datastore. If the entity does not have an ID,
    * it may be generated.
@@ -86,6 +144,20 @@ public interface DatastoreAccess {
    */
   <E> E upsert(E entity);
 
+  /**
+   * Updates or inserts the given entity in the Cloud Datastore. If the entity does not have an ID,
+   * it may be generated.
+   *
+   * @param entity
+   *          the entity to update or insert
+   * @param entityType
+   *          the entity type
+   * @return the updated/inserted entity.
+   * @throws EntityManagerException
+   *           if any error occurs while saving.
+   */
+  <E> E upsert(E entity, Type entityType);
+
   /**
    * Updates or inserts the given list of entities in the Cloud Datastore. If the entities do not
    * have a valid ID, IDs may be generated.
@@ -98,6 +170,20 @@ public interface DatastoreAccess {
    */
   <E> List<E> upsert(List<E> entities);
 
+  /**
+   * Updates or inserts the given list of entities in the Cloud Datastore. If the entities do not
+   * have a valid ID, IDs may be generated.
+   *
+   * @param entities
+   *          the entities to update/or insert.
+   * @param entityType
+   *          the entity type
+   * @return the updated or inserted entities
+   * @throws EntityManagerException
+   *           if any error occurs while saving.
+   */
+  <E> List<E> upsert(List<E> entities, Type entityType);
+
   /**
    * Deletes the given entity from the Cloud Datastore.
    * 
@@ -108,6 +194,18 @@ public interface DatastoreAccess {
    */
   void delete(Object entity);
 
+  /**
+   * Deletes the given entity from the Cloud Datastore.
+   *
+   * @param entity
+   *          the entity to delete. The entity must have it ID set for the deletion to succeed.
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if any error occurs while deleting.
+   */
+  void delete(Object entity, Type entityType);
+
   /**
    * Deletes the given entities from the Cloud Datastore.
    * 
@@ -118,37 +216,49 @@ public interface DatastoreAccess {
    */
   void delete(List<?> entities);
 
+  /**
+   * Deletes the given entities from the Cloud Datastore.
+   *
+   * @param entities
+   *          the entities to delete. The entities must have it ID set for the deletion to succeed.
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if any error occurs while deleting.
+   */
+  void delete(List<?> entities, Type entityType);
+
   /**
    * Deletes the entity with the given ID. The entity is assumed to be a root entity (no parent).
    * The entity kind will be determined from the supplied entity class.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity.
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, long id);
+  void delete(Type entityType, long id);
 
   /**
    * Deletes the entity with the given ID. The entity is assumed to be a root entity (no parent).
    * The entity kind will be determined from the supplied entity class.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity.
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, String id);
+  void delete(Type entityType, String id);
 
   /**
    * Deletes the entity with the given ID and parent key.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key
    * @param id
@@ -156,21 +266,21 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id);
+  void delete(Type entityType, DatastoreKey parentKey, long id);
 
   /**
    * Deletes the entity with the given ID and parent key.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key
    * @param id
    *          the ID of the entity.
    * @throws EntityManagerException
-   *           if any error occurs while inserting.
+   *          if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, DatastoreKey parentKey, String id);
+  void delete(Type entityType, DatastoreKey parentKey, String id);
 
   /**
    * Deletes an entity given its key.
@@ -196,8 +306,8 @@ public interface DatastoreAccess {
    * Loads and returns the entity with the given ID. The entity is assumed to be a root entity (no
    * parent). The entity kind is determined from the supplied class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity
    * @return the Entity object or <code>null</code>, if the the entity with the given ID does not
@@ -205,14 +315,14 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> E load(Class<E> entityClass, long id);
+  <E> E load(Type entityType, long id);
 
   /**
    * Loads and returns the entity with the given ID. The entity is assumed to be a root entity (no
    * parent). The entity kind is determined from the supplied class.
-   * 
-   * @param entityClass
-   *          the entity class
+   *
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity
    * @return the Entity object or <code>null</code>, if the the entity with the given ID does not
@@ -220,14 +330,14 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> E load(Class<E> entityClass, String id);
+  <E> E load(Type entityType, String id);
 
   /**
    * Loads and returns the entity with the given ID. The entity kind is determined from the supplied
    * class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key of the entity.
    * @param id
@@ -237,14 +347,14 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> E load(Class<E> entityClass, DatastoreKey parentKey, long id);
+  <E> E load(Type entityType, DatastoreKey parentKey, long id);
 
   /**
    * Loads and returns the entity with the given ID. The entity kind is determined from the supplied
    * class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key of the entity.
    * @param id
@@ -254,13 +364,13 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> E load(Class<E> entityClass, DatastoreKey parentKey, String id);
+  <E> E load(Type entityType,  DatastoreKey parentKey, String id);
 
   /**
    * Loads and returns the entity with the given key.
    * 
-   * @param entityClass
-   *          the entity class (expected result type)
+   * @param entityType
+   *          the entity type (expected result type)
    * @param key
    *          full key of the entity
    * @return the Entity object or <code>null</code>, if the the entity with the given key does not
@@ -268,14 +378,14 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while accessing the Cloud Datastore.
    */
-  <E> E load(Class<E> entityClass, DatastoreKey key);
+  <E> E load(Type entityType, DatastoreKey key);
 
   /**
    * Loads and returns the entities with the given <b>numeric IDs</b>. The entities are assumed to
    * be a root entities (no parent). The entity kind is determined from the supplied class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type.
    * @param identifiers
    *          the IDs of the entities
    * @return the list of entity objects in the same order as the given list of identifiers. If one
@@ -284,14 +394,14 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> List<E> loadById(Class<E> entityClass, List<Long> identifiers);
+  <E> List<E> loadById(Type entityType, List<Long> identifiers);
 
   /**
    * Loads and returns the entities with the given <b>names (a.k.a String IDs)</b>. The entities are
    * assumed to be root entities (no parent). The entity kind is determined from the supplied class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type.
    * @param identifiers
    *          the IDs of the entities
    * @return the list of entity objects in the same order as the given list of identifiers. If one
@@ -300,13 +410,13 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> List<E> loadByName(Class<E> entityClass, List<String> identifiers);
+  <E> List<E> loadByName(Type entityType, List<String> identifiers);
 
   /**
    * Loads and returns the entities for the given keys.
    * 
-   * @param entityClass
-   *          the entity class (expected result type)
+   * @param entityType
+   *          the entity type (expected result type)
    * @param keys
    *          entity keys to load
    * @return the Entity objects for the given keys. If one or more requested keys do not exist in
@@ -315,7 +425,7 @@ public interface DatastoreAccess {
    * @throws EntityManagerException
    *           if any error occurs while accessing the Cloud Datastore.
    */
-  <E> List<E> loadByKey(Class<E> entityClass, List<DatastoreKey> keys);
+  <E> List<E> loadByKey(Type entityType, List<DatastoreKey> keys);
 
   /**
    * Creates and returns a new {@link EntityQueryRequest} for the given GQL query string. The
@@ -362,7 +472,7 @@ public interface DatastoreAccess {
    *          the entity query request
    * @return the query response
    */
-  <E> QueryResponse<E> executeEntityQueryRequest(Class<E> expectedResultType,
+  <E> QueryResponse<E> executeEntityQueryRequest(Type expectedResultType,
       EntityQueryRequest request);
 
   /**
@@ -374,7 +484,7 @@ <E> QueryResponse<E> executeEntityQueryRequest(Class<E> expectedResultType,
    *          the projection query request
    * @return the query response
    */
-  <E> QueryResponse<E> executeProjectionQueryRequest(Class<E> expectedResultType,
+  <E> QueryResponse<E> executeProjectionQueryRequest(Type expectedResultType,
       ProjectionQueryRequest request);
 
   /**
diff --git a/src/main/java/com/jmethods/catatumbo/DatastoreBatch.java b/src/main/java/com/jmethods/catatumbo/DatastoreBatch.java
index de6278b..1d46f7e 100644
--- a/src/main/java/com/jmethods/catatumbo/DatastoreBatch.java
+++ b/src/main/java/com/jmethods/catatumbo/DatastoreBatch.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo;
 
+import java.lang.reflect.Type;
 import java.util.List;
 
 /**
@@ -41,6 +42,22 @@ public interface DatastoreBatch {
    */
   <E> E insert(E entity);
 
+  /**
+   * Adds the given entity to this batch for insertion. If the entity does not have an identifier
+   * set, ID may be automatically generated. If the same entity was added to this batch for
+   * deletion, the insert request is changed into an upsert request.
+   *
+   * @param entity
+   *          the entity to insert.
+   * @param entityType
+   *          the entity type
+   * @return the inserted entity. The inserted entity may not be same as the passed in entity. For
+   *         example, the inserted entity may contain any generated ID, key, parent key, etc.
+   * @throws EntityManagerException
+   *           if any error occurs while inserting.
+   */
+  <E> E insert(E entity, Type entityType);
+
   /**
    * Adds the given entities to this batch for insertion. If any of the entities do not have an
    * identifier set, ID may be generated automatically. If any of entities were added to this batch
@@ -55,6 +72,22 @@ public interface DatastoreBatch {
    */
   <E> List<E> insert(List<E> entities);
 
+  /**
+   * Adds the given entities to this batch for insertion. If any of the entities do not have an
+   * identifier set, ID may be generated automatically. If any of entities were added to this batch
+   * for deletion, the insert request is changed to an upsert request.
+   *
+   * @param entities
+   *          the entities to insert.
+   * @param entityType
+   *          the entity type
+   * @return the inserted entities. The inserted entities may not be same as the passed in entities.
+   *         For example, the inserted entities may contain generated ID, key, parent key, etc.
+   * @throws EntityManagerException
+   *           if any error occurs while inserting.
+   */
+  <E> List<E> insert(List<E> entities, Type entityType);
+
   /**
    * Adds the given entity to this batch for insertion. The ID allocation is deferred to the submit
    * time of this batch. Generated keys can be retrieved using
@@ -70,6 +103,23 @@ public interface DatastoreBatch {
    */
   <E> void insertWithDeferredIdAllocation(E entity);
 
+  /**
+   * Adds the given entity to this batch for insertion. The ID allocation is deferred to the submit
+   * time of this batch. Generated keys can be retrieved using
+   * {@link DatastoreBatch.Response#getGeneratedKeys()}.This method throws
+   * {@link EntityManagerException} if the entity is using a String identifier. String identifiers
+   * are allocated at add time, instead of submit time. For entities with String identifiers, use
+   * the {@link DatastoreBatch#insert(Object)} method.
+   *
+   * @param entity
+   *          the entity to insert
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entity has a String identifier or if any error occurs
+   */
+  <E> void insertWithDeferredIdAllocation(E entity, Type entityType);
+
   /**
    * Adds the given entities to this batch for insertion. The ID allocation is deferred to the
    * submit time of this batch. Generated keys can be retrieved using
@@ -85,6 +135,23 @@ public interface DatastoreBatch {
    */
   <E> void insertWithDeferredIdAllocation(List<E> entities);
 
+  /**
+   * Adds the given entities to this batch for insertion. The ID allocation is deferred to the
+   * submit time of this batch. Generated keys can be retrieved using
+   * {@link DatastoreBatch.Response#getGeneratedKeys()}.This method throws
+   * {@link EntityManagerException} if the entity is using a String identifier. String identifiers
+   * are allocated at add time, instead of submit time. For entities with String identifiers, use
+   * the {@link DatastoreBatch#insert(List)} method.
+   *
+   * @param entities
+   *          the entities to insert
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entity have a String identifier any error occurs
+   */
+  <E> void insertWithDeferredIdAllocation(List<E> entities, Type entityType);
+
   /**
    * Adds the given entity to this batch for update. The operation will fail if the entity with the
    * same key does not exist in the underlying Datastore. This method does not use optimistic
@@ -98,6 +165,21 @@ public interface DatastoreBatch {
    */
   <E> E update(E entity);
 
+  /**
+   * Adds the given entity to this batch for update. The operation will fail if the entity with the
+   * same key does not exist in the underlying Datastore. This method does not use optimistic
+   * locking even if the given entity has a field with {@link Version} annotation.
+   *
+   * @param entity
+   *          the entity to update
+   * @param entityType
+   *          the entity type
+   * @return the updated entity.
+   * @throws EntityManagerException
+   *           if any error occurs while updating.
+   */
+  <E> E update(E entity, Type entityType);
+
   /**
    * Adds the given entities to this batch for update. The operation will fail if the entities with
    * the same key do not already exist in the underlying datastore. This method does not use
@@ -112,6 +194,22 @@ public interface DatastoreBatch {
    */
   <E> List<E> update(List<E> entities);
 
+  /**
+   * Adds the given entities to this batch for update. The operation will fail if the entities with
+   * the same key do not already exist in the underlying datastore. This method does not use
+   * optimistic locking even if the given entity has a field with {@link Version} annotation.
+   *
+   *
+   * @param entities
+   *          the entities to update.
+   * @param entityType
+   *          the entity type
+   * @return the updated entities
+   * @throws EntityManagerException
+   *           if any error occurs while inserting.
+   */
+  <E> List<E> update(List<E> entities, Type entityType);
+
   /**
    * Adds the given entity to this batch for update or insert. Any prior writes to this entity in
    * this batch will be removed from this batch. If the entity does not have an ID set, ID may be
@@ -125,6 +223,21 @@ public interface DatastoreBatch {
    */
   <E> E upsert(E entity);
 
+  /**
+   * Adds the given entity to this batch for update or insert. Any prior writes to this entity in
+   * this batch will be removed from this batch. If the entity does not have an ID set, ID may be
+   * generated automatically.
+   *
+   * @param entity
+   *          the entity to update or insert
+   * @param entityType
+   *          the entity type
+   * @return the updated/inserted entity.
+   * @throws EntityManagerException
+   *           if any error occurs while saving.
+   */
+  <E> E upsert(E entity, Type entityType);
+
   /**
    * Adds the given entities to this batch for update or insert. Any prior writes of any of the
    * entities will be removed from this batch. If any of the entities do not have their IDs set, IDs
@@ -138,6 +251,21 @@ public interface DatastoreBatch {
    */
   <E> List<E> upsert(List<E> entities);
 
+  /**
+   * Adds the given entities to this batch for update or insert. Any prior writes of any of the
+   * entities will be removed from this batch. If any of the entities do not have their IDs set, IDs
+   * may be generated automatically.
+   *
+   * @param entities
+   *          the entities to update/or insert.
+   * @param entityType
+   *          the entity type
+   * @return the updated or inserted entities
+   * @throws EntityManagerException
+   *           if any error occurs while saving.
+   */
+  <E> List<E> upsert(List<E> entities, Type entityType);
+
   /**
    * Adds the given entity to this batch for insert or update. The ID (for numeric ID) allocation
    * will be deferred to the submit time of this batch. This method throws an
@@ -152,6 +280,22 @@ public interface DatastoreBatch {
    */
   <E> void upsertWithDeferredIdAllocation(E entity);
 
+  /**
+   * Adds the given entity to this batch for insert or update. The ID (for numeric ID) allocation
+   * will be deferred to the submit time of this batch. This method throws an
+   * {@link EntityManagerException} if the entity is using a String ID. Entity with String ID should
+   * use {@link DatastoreBatch#upsert(Object)} method.
+   *
+   * @param entity
+   *          the entity to update or insert
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entity has a String ID or any other error occurs while accessing the
+   *           underlying datastore.
+   */
+  <E> void upsertWithDeferredIdAllocation(E entity, Type entityType);
+
   /**
    * Adds the given entities to this batch for update or insert. The ID (for numeric ID) allocation
    * will be deferred to the submit time of this batch. This method throws an
@@ -166,6 +310,22 @@ public interface DatastoreBatch {
    */
   <E> void upsertWithDeferredIdAllocation(List<E> entities);
 
+  /**
+   * Adds the given entities to this batch for update or insert. The ID (for numeric ID) allocation
+   * will be deferred to the submit time of this batch. This method throws an
+   * {@link EntityManagerException} if the entities have String identifiers. Entities with String
+   * identifiers should use {@link DatastoreBatch#upsert(List)} method.
+   *
+   * @param entities
+   *          the entities to update or insert
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entities have String identifiers or any other error occurs while accessing the
+   *           underlying datastore.
+   */
+  <E> void upsertWithDeferredIdAllocation(List<E> entities, Type entityType);
+
   /**
    * Adds the given entity to this batch for deletion.
    * 
@@ -176,6 +336,18 @@ public interface DatastoreBatch {
    */
   void delete(Object entity);
 
+  /**
+   * Adds the given entity to this batch for deletion.
+   *
+   * @param entity
+   *          the entity to delete.
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if any error occurs while deleting.
+   */
+  void delete(Object entity, Type entityType);
+
   /**
    * Adds the given entities to this batch for deletion.
    * 
@@ -186,35 +358,47 @@ public interface DatastoreBatch {
    */
   void delete(List<?> entities);
 
+  /**
+   * Adds the given entities to this batch for deletion.
+   *
+   * @param entities
+   *          the entities to delete.
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if any error occurs while deleting.
+   */
+  void delete(List<?> entities, Type entityType);
+
   /**
    * Adds the given ID and entity type to this batch for deletion.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity.
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, long id);
+  <E> void delete(Type entityType, long id);
 
   /**
    * Adds the given ID and entity type to this batch for deletion.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity.
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, String id);
+  <E> void delete(Type entityType, String id);
 
   /**
    * Adds the given entity type, parent key and ID to this batch for deletion.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key
    * @param id
@@ -222,13 +406,13 @@ public interface DatastoreBatch {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id);
+  <E> void delete(Type entityType, DatastoreKey parentKey, long id);
 
   /**
    * Adds the given entity type, parent key and ID to this batch for deletion.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key
    * @param id
@@ -236,7 +420,7 @@ public interface DatastoreBatch {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  <E> void delete(Class<E> entityClass, DatastoreKey parentKey, String id);
+  <E> void delete(Type entityType, DatastoreKey parentKey, String id);
 
   /**
    * Adds the given key to this batch for deletion.
@@ -250,7 +434,7 @@ public interface DatastoreBatch {
 
   /**
    * Adds the given keys to this batch for deletion.
-   * 
+   *
    * @param keys
    *          the keys to delete
    * @throws EntityManagerException
diff --git a/src/main/java/com/jmethods/catatumbo/DatastoreTransaction.java b/src/main/java/com/jmethods/catatumbo/DatastoreTransaction.java
index 874fe39..767f8ca 100644
--- a/src/main/java/com/jmethods/catatumbo/DatastoreTransaction.java
+++ b/src/main/java/com/jmethods/catatumbo/DatastoreTransaction.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo;
 
+import java.lang.reflect.Type;
 import java.util.List;
 
 /**
@@ -43,6 +44,20 @@ public interface DatastoreTransaction extends DatastoreAccess {
    */
   <E> void insertWithDeferredIdAllocation(E entity);
 
+  /**
+   * Inserts the given entity. ID allocation is deferred to the submit time. Generated key, if any,
+   * can be retrieved by calling {@link DatastoreTransaction.Response#getGeneratedKeys()}.
+   *
+   * @param entity
+   *          the entity to insert.
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entity has a String Identifier or any other error occurs while accessing the
+   *           underlying Datastore.
+   */
+  <E> void insertWithDeferredIdAllocation(E entity, Type entityType);
+
   /**
    * Inserts the given entities. ID allocation is deferred to the submit time. Generated keys, if
    * any, can be retrieved by calling {@link DatastoreTransaction.Response#getGeneratedKeys()}.
@@ -55,6 +70,20 @@ public interface DatastoreTransaction extends DatastoreAccess {
    */
   <E> void insertWithDeferredIdAllocation(List<E> entities);
 
+  /**
+   * Inserts the given entities. ID allocation is deferred to the submit time. Generated keys, if
+   * any, can be retrieved by calling {@link DatastoreTransaction.Response#getGeneratedKeys()}.
+   *
+   * @param entities
+   *          the entities to insert
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entity has a String Identifier or any other error occurs while accessing the
+   *           underlying Datastore.
+   */
+  <E> void insertWithDeferredIdAllocation(List<E> entities, Type entityType);
+
   /**
    * Updates or inserts the given entity. ID allocation is deferred to the submit time. Generated
    * key, if any, can be retrieved by calling
@@ -69,6 +98,22 @@ public interface DatastoreTransaction extends DatastoreAccess {
    */
   <E> void upsertWithDeferredIdAllocation(E entity);
 
+  /**
+   * Updates or inserts the given entity. ID allocation is deferred to the submit time. Generated
+   * key, if any, can be retrieved by calling
+   * {@link DatastoreTransaction.Response#getGeneratedKeys()}.
+   *
+   *
+   * @param entity
+   *          the entity to update or insert.
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entity has a String Identifier or any other error occurs while accessing the
+   *           underlying Datastore.
+   */
+  <E> void upsertWithDeferredIdAllocation(E entity, Type entityType);
+
   /**
    * Updates or Inserts the given entities. ID allocation is deferred to the submit time. Generated
    * keys, if any, can be retrieved by calling
@@ -83,6 +128,22 @@ public interface DatastoreTransaction extends DatastoreAccess {
    */
   <E> void upsertWithDeferredIdAllocation(List<E> entities);
 
+  /**
+   * Updates or Inserts the given entities. ID allocation is deferred to the submit time. Generated
+   * keys, if any, can be retrieved by calling
+   * {@link DatastoreTransaction.Response#getGeneratedKeys()}.
+   *
+   *
+   * @param entities
+   *          the entities to update or insert
+   * @param entityType
+   *          the entity type
+   * @throws EntityManagerException
+   *           if the entity has a String Identifier or any other error occurs while accessing the
+   *           underlying Datastore.
+   */
+  <E> void upsertWithDeferredIdAllocation(List<E> entities, Type entityType);
+
   /**
    * Tells if this DatastoreTransaction is still active.
    * 
diff --git a/src/main/java/com/jmethods/catatumbo/EntityManager.java b/src/main/java/com/jmethods/catatumbo/EntityManager.java
index f1f1ffb..da79db7 100644
--- a/src/main/java/com/jmethods/catatumbo/EntityManager.java
+++ b/src/main/java/com/jmethods/catatumbo/EntityManager.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo;
 
+import java.lang.reflect.Type;
 import java.util.List;
 
 /**
@@ -30,13 +31,13 @@ public interface EntityManager extends DatastoreAccess {
   /**
    * Deletes all entities of given Kind.
    * 
-   * @param entityClass
-   *          the entity class - The entity Kind will be determined from this class.
+   * @param entityType
+   *          the entity type - The entity Kind will be determined from this class.
    * @return the number of entities that were deleted
    * @throws EntityManagerException
    *           if any error occurs while deleting.
    */
-  <E> long deleteAll(Class<E> entityClass);
+  <E> long deleteAll(Type entityType);
 
   /**
    * Deletes all entities of given Kind.
@@ -120,6 +121,23 @@ public interface EntityManager extends DatastoreAccess {
    */
   List<DatastoreKey> allocateId(List<Object> entities);
 
+  /**
+   * Allocates IDs for the given entities and returns the allocated IDs. Each entity in the list
+   * must have a its identifier of type numeric (long/Long).
+   *
+   * @param entities
+   *          the entities
+   * @param entityType
+   *          the entity type
+   * @return a list of {@link DatastoreKey}s.
+   * @throws IllegalArgumentException
+   *           if any of the entities in the list do not have a numeric ID type or a valid ID is
+   *           already set.
+   * @throws EntityManagerException
+   *           if any error occurs during key allocation
+   */
+  List<DatastoreKey> allocateId(List<Object> entities, Type entityType);
+
   /**
    * Allocates ID for the given entity and returns the allocated ID. The entity must have its
    * identifier of type numeric (long/Long).
@@ -135,4 +153,21 @@ public interface EntityManager extends DatastoreAccess {
    */
   DatastoreKey allocateId(Object entity);
 
+  /**
+   * Allocates ID for the given entity and returns the allocated ID. The entity must have its
+   * identifier of type numeric (long/Long).
+   *
+   * @param entity
+   *          the the entity.
+   * @param entityType
+   *          the entity type
+   * @return the allocated ID {@link DatastoreKey}.
+   * @throws IllegalArgumentException
+   *           if the ID type of the entity is not numeric, or if the entity has a valid ID
+   *           (non-null and non-zero).
+   * @throws EntityManagerException
+   *           if any error occurs during ID allocation
+   */
+  DatastoreKey allocateId(Object entity, Type entityType);
+
 }
diff --git a/src/main/java/com/jmethods/catatumbo/MapperFactory.java b/src/main/java/com/jmethods/catatumbo/MapperFactory.java
index 0c26878..3af6c63 100644
--- a/src/main/java/com/jmethods/catatumbo/MapperFactory.java
+++ b/src/main/java/com/jmethods/catatumbo/MapperFactory.java
@@ -111,14 +111,15 @@ public static MapperFactory getInstance() {
    * 
    * @param field
    *          the field
+   * @param fieldType
+   *          the field type
    * @return the mapper for the given field.
    */
-  public Mapper getMapper(Field field) {
+  public Mapper getMapper(Field field, Class<?> fieldType) {
     PropertyMapper propertyMapperAnnotation = field.getAnnotation(PropertyMapper.class);
     if (propertyMapperAnnotation != null) {
       return createCustomMapper(field, propertyMapperAnnotation);
     }
-    Class<?> fieldType = field.getType();
     if (fieldType.equals(BigDecimal.class)) {
       Decimal decimalAnnotation = field.getAnnotation(Decimal.class);
       if (decimalAnnotation != null) {
@@ -128,7 +129,10 @@ public Mapper getMapper(Field field) {
     if (List.class.isAssignableFrom(fieldType) || Set.class.isAssignableFrom(fieldType)) {
       return CollectionMapperFactory.getInstance().getMapper(field);
     }
-    return getMapper(field.getGenericType());
+    if (Map.class.isAssignableFrom(fieldType)) {
+      return getMapper(field.getGenericType());
+    }
+    return getMapper(fieldType);
   }
 
   /**
diff --git a/src/main/java/com/jmethods/catatumbo/impl/DatastoreUtils.java b/src/main/java/com/jmethods/catatumbo/impl/DatastoreUtils.java
index 3a58c8c..777151f 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/DatastoreUtils.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/DatastoreUtils.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo.impl;
 
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -63,19 +64,19 @@ private DatastoreUtils() {
    * Converts the given list of native entities to a list of model objects of given type,
    * <code>entityClass</code>.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param nativeEntities
    *          native entities to convert
    * @return the list of model objects
    */
-  static <E> List<E> toEntities(Class<E> entityClass, List<Entity> nativeEntities) {
+  public static <E> List<E> toEntities(Type entityType, List<Entity> nativeEntities) {
     if (nativeEntities == null || nativeEntities.isEmpty()) {
       return new ArrayList<>();
     }
     List<E> entities = new ArrayList<>(nativeEntities.size());
     for (Entity nativeEntity : nativeEntities) {
-      E entity = Unmarshaller.unmarshal(nativeEntity, entityClass);
+      E entity = Unmarshaller.unmarshal(nativeEntity, entityType);
       entities.add(entity);
     }
     return entities;
@@ -85,17 +86,17 @@ static <E> List<E> toEntities(Class<E> entityClass, List<Entity> nativeEntities)
    * Converts the given array of native entities to a list of model objects of given type,
    * <code>entityClass</code>.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param nativeEntities
    *          native entities to convert
    * @return the list of model objects
    */
-  static <E> List<E> toEntities(Class<E> entityClass, Entity[] nativeEntities) {
+  public static <E> List<E> toEntities(Type entityType, Entity[] nativeEntities) {
     if (nativeEntities == null || nativeEntities.length == 0) {
       return new ArrayList<>();
     }
-    return toEntities(entityClass, Arrays.asList(nativeEntities));
+    return toEntities(entityType, Arrays.asList(nativeEntities));
   }
 
   /**
@@ -110,11 +111,11 @@ static <E> List<E> toEntities(Class<E> entityClass, Entity[] nativeEntities) {
    * @return the equivalent FullEntity array
    */
   static FullEntity<?>[] toNativeFullEntities(List<?> entities, DefaultEntityManager entityManager,
-      Marshaller.Intent intent) {
+      Marshaller.Intent intent, Type entityType) {
     FullEntity<?>[] nativeEntities = new FullEntity[entities.size()];
     for (int i = 0; i < entities.size(); i++) {
       nativeEntities[i] = (FullEntity<?>) Marshaller.marshal(entityManager, entities.get(i),
-          intent);
+              intent, entityType);
     }
     return nativeEntities;
   }
@@ -131,10 +132,11 @@ static FullEntity<?>[] toNativeFullEntities(List<?> entities, DefaultEntityManag
    * @return the equivalent Entity array
    */
   static Entity[] toNativeEntities(List<?> entities, DefaultEntityManager entityManager,
-      Marshaller.Intent intent) {
+                                   Marshaller.Intent intent, Type entityType) {
     Entity[] nativeEntities = new Entity[entities.size()];
     for (int i = 0; i < entities.size(); i++) {
-      nativeEntities[i] = (Entity) Marshaller.marshal(entityManager, entities.get(i), intent);
+      nativeEntities[i] = (Entity) Marshaller.marshal(entityManager, entities.get(i),
+              intent, entityType);
     }
     return nativeEntities;
   }
@@ -213,8 +215,8 @@ static Key[] toNativeKeys(List<DatastoreKey> keys) {
    * @throws EntityManagerException
    *           if the given entity does not use a numeric ID
    */
-  static void validateDeferredIdAllocation(Object entity) {
-    IdentifierMetadata identifierMetadata = EntityIntrospector.getIdentifierMetadata(entity);
+  public static void validateDeferredIdAllocation(Object entity, Type type) {
+    IdentifierMetadata identifierMetadata = EntityIntrospector.getIdentifierMetadata(entity, type);
     if (identifierMetadata.getDataType() == DataType.STRING) {
       throw new EntityManagerException(
           "Deferred ID allocation is not applicable for entities with String identifiers. ");
diff --git a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreBatch.java b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreBatch.java
index eefe054..4585f03 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreBatch.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreBatch.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo.impl;
 
+import java.lang.reflect.Type;
 import java.util.List;
 
 import com.google.cloud.datastore.Batch;
@@ -87,20 +88,35 @@ public Batch getNativeBatch() {
 
   @Override
   public <E> E insert(E entity) {
-    return writer.insert(entity);
+    return insert(entity, null);
+  }
+
+  @Override
+  public <E> E insert(E entity, Type entityType) {
+    return writer.insert(entity, entityType);
   }
 
   @Override
   public <E> List<E> insert(List<E> entities) {
-    return writer.insert(entities);
+    return insert(entities, null);
+  }
+
+  @Override
+  public <E> List<E> insert(List<E> entities, Type entityType) {
+    return writer.insert(entities, entityType);
   }
 
   @Override
   public <E> void insertWithDeferredIdAllocation(E entity) {
+    insertWithDeferredIdAllocation(entity, null);
+  }
+
+  @Override
+  public <E> void insertWithDeferredIdAllocation(E entity, Type entityType) {
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entity);
+      DatastoreUtils.validateDeferredIdAllocation(entity, entityType);
       FullEntity<?> nativeEntity = (FullEntity<?>) Marshaller.marshal(entityManager, entity,
-          Intent.INSERT);
+              Intent.INSERT, entityType);
       nativeBatch.addWithDeferredIdAllocation(nativeEntity);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -109,13 +125,18 @@ public <E> void insertWithDeferredIdAllocation(E entity) {
 
   @Override
   public <E> void insertWithDeferredIdAllocation(List<E> entities) {
+    insertWithDeferredIdAllocation(entities, null);
+  }
+
+  @Override
+  public <E> void insertWithDeferredIdAllocation(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return;
     }
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entities.get(0));
+      DatastoreUtils.validateDeferredIdAllocation(entities.get(0), entityType);
       FullEntity<?>[] nativeEntities = DatastoreUtils.toNativeFullEntities(entities, entityManager,
-          Intent.INSERT);
+          Intent.INSERT, entityType);
       nativeBatch.addWithDeferredIdAllocation(nativeEntities);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -125,30 +146,55 @@ public <E> void insertWithDeferredIdAllocation(List<E> entities) {
 
   @Override
   public <E> E update(E entity) {
-    return writer.update(entity);
+    return update(entity, null);
+  }
+
+  @Override
+  public <E> E update(E entity, Type entityType) {
+    return writer.update(entity, entityType);
   }
 
   @Override
   public <E> List<E> update(List<E> entities) {
-    return writer.update(entities);
+    return update(entities, null);
+  }
+
+  @Override
+  public <E> List<E> update(List<E> entities, Type entityType) {
+    return writer.update(entities, entityType);
   }
 
   @Override
   public <E> E upsert(E entity) {
-    return writer.upsert(entity);
+    return upsert(entity, null);
+  }
+
+  @Override
+  public <E> E upsert(E entity, Type entityType) {
+    return writer.upsert(entity, entityType);
   }
 
   @Override
   public <E> List<E> upsert(List<E> entities) {
-    return writer.upsert(entities);
+    return upsert(entities, null);
+  }
+
+  @Override
+  public <E> List<E> upsert(List<E> entities, Type entityType) {
+    return writer.upsert(entities, entityType);
   }
 
   @Override
   public <E> void upsertWithDeferredIdAllocation(E entity) {
+    upsertWithDeferredIdAllocation(entity, null);
+  }
+
+  @Override
+  public <E> void upsertWithDeferredIdAllocation(E entity, Type entityType) {
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entity);
+      DatastoreUtils.validateDeferredIdAllocation(entity, entityType);
       FullEntity<?> nativeEntity = (FullEntity<?>) Marshaller.marshal(entityManager, entity,
-          Intent.UPSERT);
+              Intent.UPSERT, entityType);
       nativeBatch.putWithDeferredIdAllocation(nativeEntity);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -158,13 +204,18 @@ public <E> void upsertWithDeferredIdAllocation(E entity) {
 
   @Override
   public <E> void upsertWithDeferredIdAllocation(List<E> entities) {
+    upsertWithDeferredIdAllocation(entities, null);
+  }
+
+  @Override
+  public <E> void upsertWithDeferredIdAllocation(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return;
     }
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entities.get(0));
+      DatastoreUtils.validateDeferredIdAllocation(entities.get(0), entityType);
       FullEntity<?>[] nativeEntities = DatastoreUtils.toNativeFullEntities(entities, entityManager,
-          Intent.UPSERT);
+          Intent.UPSERT, entityType);
       nativeBatch.putWithDeferredIdAllocation(nativeEntities);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -174,32 +225,42 @@ public <E> void upsertWithDeferredIdAllocation(List<E> entities) {
 
   @Override
   public void delete(Object entity) {
-    writer.delete(entity);
+    delete(entity, null);
+  }
+
+  @Override
+  public void delete(Object entity, Type entityType) {
+    writer.delete(entity, entityType);
   }
 
   @Override
   public void delete(List<?> entities) {
-    writer.delete(entities);
+    delete(entities, null);
+  }
+
+  @Override
+  public void delete(List<?> entities, Type entityType) {
+    writer.delete(entities, entityType);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, long id) {
-    writer.delete(entityClass, id);
+  public <E> void delete(Type entityType, long id) {
+    writer.delete(entityType, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, String id) {
-    writer.delete(entityClass, id);
+  public <E> void delete(Type entityType, String id) {
+    writer.delete(entityType, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id) {
-    writer.delete(entityClass, parentKey, id);
+  public <E> void delete(Type entityType, DatastoreKey parentKey, long id) {
+    writer.delete(entityType, parentKey, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, String id) {
-    writer.delete(entityClass, parentKey, id);
+  public <E> void delete(Type entityType, DatastoreKey parentKey, String id) {
+    writer.delete(entityType, parentKey, id);
   }
 
   @Override
diff --git a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreReader.java b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreReader.java
index 8aed460..078c155 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreReader.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreReader.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo.impl;
 
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -92,8 +93,8 @@ public DefaultDatastoreReader(DefaultDatastoreTransaction transaction) {
    * Loads and returns the entity with the given ID. The entity is assumed to be a root entity (no
    * parent). The entity kind is determined from the supplied class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param id
    *          the ID of the entity
    * @return the Entity object or <code>null</code>, if the the entity with the given ID does not
@@ -101,16 +102,16 @@ public DefaultDatastoreReader(DefaultDatastoreTransaction transaction) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> E load(Class<E> entityClass, long id) {
-    return load(entityClass, null, id);
+  public <E> E load(Type entityType, long id) {
+    return load(entityType, null, id);
   }
 
   /**
    * Loads and returns the entity with the given ID. The entity kind is determined from the supplied
    * class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param parentKey
    *          the parent key of the entity.
    * @param id
@@ -120,23 +121,23 @@ public <E> E load(Class<E> entityClass, long id) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> E load(Class<E> entityClass, DatastoreKey parentKey, long id) {
-    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+  public <E> E load(Type entityType, DatastoreKey parentKey, long id) {
+    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
     Key nativeKey;
     if (parentKey == null) {
       nativeKey = entityManager.newNativeKeyFactory().setKind(entityMetadata.getKind()).newKey(id);
     } else {
       nativeKey = Key.newBuilder(parentKey.nativeKey(), entityMetadata.getKind(), id).build();
     }
-    return fetch(entityClass, nativeKey);
+    return fetch(entityType, nativeKey);
   }
 
   /**
    * Loads and returns the entity with the given ID. The entity is assumed to be a root entity (no
    * parent). The entity kind is determined from the supplied class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param id
    *          the ID of the entity
    * @return the Entity object or <code>null</code>, if the the entity with the given ID does not
@@ -144,16 +145,16 @@ public <E> E load(Class<E> entityClass, DatastoreKey parentKey, long id) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> E load(Class<E> entityClass, String id) {
-    return load(entityClass, null, id);
+  public <E> E load(Type entityType, String id) {
+    return load(entityType, null, id);
   }
 
   /**
    * Loads and returns the entity with the given ID. The entity kind is determined from the supplied
    * class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param parentKey
    *          the parent key of the entity.
    * @param id
@@ -163,21 +164,21 @@ public <E> E load(Class<E> entityClass, String id) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> E load(Class<E> entityClass, DatastoreKey parentKey, String id) {
-    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+  public <E> E load(Type entityType, DatastoreKey parentKey, String id) {
+    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
     Key nativeKey;
     if (parentKey == null) {
       nativeKey = entityManager.newNativeKeyFactory().setKind(entityMetadata.getKind()).newKey(id);
     } else {
       nativeKey = Key.newBuilder(parentKey.nativeKey(), entityMetadata.getKind(), id).build();
     }
-    return fetch(entityClass, nativeKey);
+    return fetch(entityType, nativeKey);
   }
 
   /**
    * Retrieves and returns the entity with the given key.
    * 
-   * @param entityClass
+   * @param entityType
    *          the expected result type
    * @param key
    *          the entity key
@@ -186,16 +187,16 @@ public <E> E load(Class<E> entityClass, DatastoreKey parentKey, String id) {
    * @throws EntityManagerException
    *           if any error occurs while accessing the Datastore.
    */
-  public <E> E load(Class<E> entityClass, DatastoreKey key) {
-    return fetch(entityClass, key.nativeKey());
+  public <E> E load(Type entityType, DatastoreKey key) {
+    return fetch(entityType, key.nativeKey());
   }
 
   /**
    * Loads and returns the entities with the given <b>numeric IDs</b>. The entities are assumed to
    * be a root entities (no parent). The entity kind is determined from the supplied class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param identifiers
    *          the IDs of the entities
    * @return the list of entity objects in the same order as the given list of identifiers. If one
@@ -204,17 +205,17 @@ public <E> E load(Class<E> entityClass, DatastoreKey key) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> List<E> loadById(Class<E> entityClass, List<Long> identifiers) {
-    Key[] nativeKeys = longListToNativeKeys(entityClass, identifiers);
-    return fetch(entityClass, nativeKeys);
+  public <E> List<E> loadById(Type entityType, List<Long> identifiers) {
+    Key[] nativeKeys = longListToNativeKeys(entityType, identifiers);
+    return fetch(entityType, nativeKeys);
   }
 
   /**
    * Loads and returns the entities with the given <b>names (a.k.a String IDs)</b>. The entities are
    * assumed to be root entities (no parent). The entity kind is determined from the supplied class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @param identifiers
    *          the IDs of the entities
    * @return the list of entity objects in the same order as the given list of identifiers. If one
@@ -223,16 +224,16 @@ public <E> List<E> loadById(Class<E> entityClass, List<Long> identifiers) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> List<E> loadByName(Class<E> entityClass, List<String> identifiers) {
-    Key[] nativeKeys = stringListToNativeKeys(entityClass, identifiers);
-    return fetch(entityClass, nativeKeys);
+  public <E> List<E> loadByName(Type entityType, List<String> identifiers) {
+    Key[] nativeKeys = stringListToNativeKeys(entityType, identifiers);
+    return fetch(entityType, nativeKeys);
   }
 
   /**
    * Retrieves and returns the entities for the given keys.
    * 
-   * @param entityClass
-   *          the expected result type
+   * @param entityType
+   *          the entity type
    * @param keys
    *          the entity keys
    * @return the entities for the given keys. If one or more requested keys do not exist in the
@@ -241,26 +242,26 @@ public <E> List<E> loadByName(Class<E> entityClass, List<String> identifiers) {
    * @throws EntityManagerException
    *           if any error occurs while accessing the Datastore.
    */
-  public <E> List<E> loadByKey(Class<E> entityClass, List<DatastoreKey> keys) {
+  public <E> List<E> loadByKey(Type entityType, List<DatastoreKey> keys) {
     Key[] nativeKeys = DatastoreUtils.toNativeKeys(keys);
-    return fetch(entityClass, nativeKeys);
+    return fetch(entityType, nativeKeys);
   }
 
   /**
    * Fetches the entity given the native key.
    * 
-   * @param entityClass
+   * @param entityType
    *          the expected result type
    * @param nativeKey
    *          the native key
    * @return the entity with the given key, or <code>null</code>, if no entity exists with the given
    *         key.
    */
-  private <E> E fetch(Class<E> entityClass, Key nativeKey) {
+  private <E> E fetch(Type entityType, Key nativeKey) {
     try {
       Entity nativeEntity = nativeReader.get(nativeKey);
-      E entity = Unmarshaller.unmarshal(nativeEntity, entityClass);
-      entityManager.executeEntityListeners(CallbackType.POST_LOAD, entity);
+      E entity = Unmarshaller.unmarshal(nativeEntity, entityType);
+      entityManager.executeEntityListeners(CallbackType.POST_LOAD, entity, entityType);
       return entity;
     } catch (DatastoreException exp) {
       throw new EntityManagerException(exp);
@@ -270,18 +271,18 @@ private <E> E fetch(Class<E> entityClass, Key nativeKey) {
   /**
    * Fetches a list of entities for the given native keys.
    * 
-   * @param entityClass
+   * @param entityType
    *          the expected result type
    * @param nativeKeys
    *          the native keys of the entities
    * @return the list of entities. If one or more keys do not exist, the corresponding item in the
    *         returned list will be <code>null</code>.
    */
-  private <E> List<E> fetch(Class<E> entityClass, Key[] nativeKeys) {
+  private <E> List<E> fetch(Type entityType, Key[] nativeKeys) {
     try {
       List<Entity> nativeEntities = nativeReader.fetch(nativeKeys);
-      List<E> entities = DatastoreUtils.toEntities(entityClass, nativeEntities);
-      entityManager.executeEntityListeners(CallbackType.POST_LOAD, entities);
+      List<E> entities = DatastoreUtils.toEntities(entityType, nativeEntities);
+      entityManager.executeEntityListeners(CallbackType.POST_LOAD, entities, entityType);
       return entities;
     } catch (DatastoreException exp) {
       throw new EntityManagerException(exp);
@@ -339,8 +340,8 @@ public KeyQueryRequest createKeyQueryRequest(String query) {
    *          the entity query request
    * @return the query response
    */
-  public <E> QueryResponse<E> executeEntityQueryRequest(Class<E> expectedResultType,
-      EntityQueryRequest request) {
+  public <E> QueryResponse<E> executeEntityQueryRequest(Type expectedResultType,
+                                                        EntityQueryRequest request) {
     try {
       GqlQuery.Builder<Entity> queryBuilder = Query.newGqlQueryBuilder(ResultType.ENTITY,
           request.getQuery());
@@ -360,7 +361,7 @@ public <E> QueryResponse<E> executeEntityQueryRequest(Class<E> expectedResultTyp
       }
       response.setResults(entities);
       response.setEndCursor(new DefaultDatastoreCursor(results.getCursorAfter().toUrlSafe()));
-      entityManager.executeEntityListeners(CallbackType.POST_LOAD, entities);
+      entityManager.executeEntityListeners(CallbackType.POST_LOAD, entities, expectedResultType);
       return response;
     } catch (DatastoreException exp) {
       throw new EntityManagerException(exp);
@@ -376,7 +377,7 @@ public <E> QueryResponse<E> executeEntityQueryRequest(Class<E> expectedResultTyp
    *          the projection query request
    * @return the query response
    */
-  public <E> QueryResponse<E> executeProjectionQueryRequest(Class<E> expectedResultType,
+  public <E> QueryResponse<E> executeProjectionQueryRequest(Type expectedResultType,
       ProjectionQueryRequest request) {
     try {
       GqlQuery.Builder<ProjectionEntity> queryBuilder = Query
@@ -440,17 +441,17 @@ public QueryResponse<DatastoreKey> executeKeyQueryRequest(KeyQueryRequest reques
   /**
    * Converts the given list of identifiers into an array of native Key objects.
    * 
-   * @param entityClass
+   * @param entityType
    *          the entity class to which these identifiers belong to.
    * @param identifiers
    *          the list of identifiers to convert.
    * @return an array of Key objects
    */
-  private Key[] longListToNativeKeys(Class<?> entityClass, List<Long> identifiers) {
+  private Key[] longListToNativeKeys(Type entityType, List<Long> identifiers) {
     if (identifiers == null || identifiers.isEmpty()) {
       return new Key[0];
     }
-    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
     Key[] nativeKeys = new Key[identifiers.size()];
     KeyFactory keyFactory = entityManager.newNativeKeyFactory();
     keyFactory.setKind(entityMetadata.getKind());
@@ -463,18 +464,18 @@ private Key[] longListToNativeKeys(Class<?> entityClass, List<Long> identifiers)
 
   /**
    * Converts the given list of identifiers into an array of native Key objects.
-   * 
-   * @param entityClass
+   *
+   * @param entityType
    *          the entity class to which these identifiers belong to.
    * @param identifiers
    *          the list of identifiers to convert.
    * @return an array of Key objects
    */
-  private Key[] stringListToNativeKeys(Class<?> entityClass, List<String> identifiers) {
+  private Key[] stringListToNativeKeys(Type entityType, List<String> identifiers) {
     if (identifiers == null || identifiers.isEmpty()) {
       return new Key[0];
     }
-    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
     Key[] nativeKeys = new Key[identifiers.size()];
     KeyFactory keyFactory = entityManager.newNativeKeyFactory();
     keyFactory.setKind(entityMetadata.getKind());
diff --git a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreTransaction.java b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreTransaction.java
index 38db571..b320fa8 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreTransaction.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreTransaction.java
@@ -18,6 +18,7 @@
 
 import static com.jmethods.catatumbo.impl.DatastoreUtils.toNativeFullEntities;
 
+import java.lang.reflect.Type;
 import java.util.List;
 
 import com.google.cloud.datastore.Datastore;
@@ -100,10 +101,15 @@ public Transaction getNativeTransaction() {
 
   @Override
   public <E> void insertWithDeferredIdAllocation(E entity) {
+    insertWithDeferredIdAllocation(entity, null);
+  }
+
+  @Override
+  public <E> void insertWithDeferredIdAllocation(E entity, Type entityType) {
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entity);
+      DatastoreUtils.validateDeferredIdAllocation(entity, entityType);
       FullEntity<?> nativeEntity = (FullEntity<?>) Marshaller.marshal(entityManager, entity,
-          Intent.INSERT);
+              Intent.INSERT, entityType);
       nativeTransaction.addWithDeferredIdAllocation(nativeEntity);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -113,12 +119,18 @@ public <E> void insertWithDeferredIdAllocation(E entity) {
 
   @Override
   public <E> void insertWithDeferredIdAllocation(List<E> entities) {
+    insertWithDeferredIdAllocation(entities, null);
+  }
+
+  @Override
+  public <E> void insertWithDeferredIdAllocation(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return;
     }
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entities.get(0));
-      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager, Intent.INSERT);
+      DatastoreUtils.validateDeferredIdAllocation(entities.get(0), entityType);
+      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager,
+              Intent.INSERT, entityType);
       nativeTransaction.addWithDeferredIdAllocation(nativeEntities);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -127,10 +139,15 @@ public <E> void insertWithDeferredIdAllocation(List<E> entities) {
 
   @Override
   public <E> void upsertWithDeferredIdAllocation(E entity) {
+    upsertWithDeferredIdAllocation(entity, null);
+  }
+
+  @Override
+  public <E> void upsertWithDeferredIdAllocation(E entity, Type entityType) {
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entity);
+      DatastoreUtils.validateDeferredIdAllocation(entity, entityType);
       FullEntity<?> nativeEntity = (FullEntity<?>) Marshaller.marshal(entityManager, entity,
-          Intent.UPSERT);
+              Intent.UPSERT, entityType);
       nativeTransaction.putWithDeferredIdAllocation(nativeEntity);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -140,12 +157,18 @@ public <E> void upsertWithDeferredIdAllocation(E entity) {
 
   @Override
   public <E> void upsertWithDeferredIdAllocation(List<E> entities) {
+    upsertWithDeferredIdAllocation(entities, null);
+  }
+
+  @Override
+  public <E> void upsertWithDeferredIdAllocation(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return;
     }
     try {
-      DatastoreUtils.validateDeferredIdAllocation(entities.get(0));
-      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager, Intent.UPSERT);
+      DatastoreUtils.validateDeferredIdAllocation(entities.get(0), entityType);
+      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager,
+              Intent.UPSERT, entityType);
       nativeTransaction.putWithDeferredIdAllocation(nativeEntities);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -217,62 +240,102 @@ public List<DatastoreKey> getGeneratedKeys() {
 
   @Override
   public <E> E insert(E entity) {
-    return writer.insert(entity);
+    return insert(entity, null);
+  }
+
+  @Override
+  public <E> E insert(E entity, Type entityType) {
+    return writer.insert(entity, entityType);
   }
 
   @Override
   public <E> List<E> insert(List<E> entities) {
-    return writer.insert(entities);
+    return insert(entities, null);
+  }
+
+  @Override
+  public <E> List<E> insert(List<E> entities, Type entityType) {
+    return writer.insert(entities, entityType);
   }
 
   @Override
   public <E> E update(E entity) {
-    return writer.updateWithOptimisticLock(entity);
+    return update(entity, null);
+  }
+
+  @Override
+  public <E> E update(E entity, Type entityType) {
+    return writer.updateWithOptimisticLock(entity, entityType);
   }
 
   @Override
   public <E> List<E> update(List<E> entities) {
-    return writer.updateWithOptimisticLock(entities);
+    return update(entities, null);
+  }
+
+  @Override
+  public <E> List<E> update(List<E> entities, Type entityType) {
+    return writer.updateWithOptimisticLock(entities, entityType);
   }
 
   @Override
   public <E> E upsert(E entity) {
-    return writer.upsert(entity);
+    return upsert(entity, null);
+  }
+
+  @Override
+  public <E> E upsert(E entity, Type entityType) {
+    return writer.upsert(entity, entityType);
   }
 
   @Override
   public <E> List<E> upsert(List<E> entities) {
-    return writer.upsert(entities);
+    return upsert(entities, null);
+  }
+
+  @Override
+  public <E> List<E> upsert(List<E> entities, Type entityType) {
+    return writer.upsert(entities, entityType);
   }
 
   @Override
   public void delete(Object entity) {
-    writer.delete(entity);
+    delete(entity, null);
+  }
+
+  @Override
+  public void delete(Object entity, Type entityType) {
+    writer.delete(entity, entityType);
   }
 
   @Override
   public void delete(List<?> entities) {
-    writer.delete(entities);
+    delete(entities, null);
+  }
+
+  @Override
+  public void delete(List<?> entities, Type entityType) {
+    writer.delete(entities, entityType);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, long id) {
-    writer.delete(entityClass, id);
+  public void delete(Type entityType, long id) {
+    writer.delete(entityType, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, String id) {
-    writer.delete(entityClass, id);
+  public void delete(Type entityType, String id) {
+    writer.delete(entityType, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id) {
-    writer.delete(entityClass, parentKey, id);
+  public void delete(Type entityType, DatastoreKey parentKey, long id) {
+    writer.delete(entityType, parentKey, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, String id) {
-    writer.delete(entityClass, parentKey, id);
+  public void delete(Type entityType, DatastoreKey parentKey, String id) {
+    writer.delete(entityType, parentKey, id);
   }
 
   @Override
@@ -286,43 +349,43 @@ public void deleteByKey(List<DatastoreKey> keys) {
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, long id) {
-    return reader.load(entityClass, id);
+  public <E> E load(Type entityType, long id) {
+    return reader.load(entityType, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, String id) {
-    return reader.load(entityClass, id);
+  public <E> E load(Type entityType, String id) {
+    return reader.load(entityType, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, DatastoreKey parentKey, long id) {
-    return reader.load(entityClass, parentKey, id);
+  public <E> E load(Type entityType, DatastoreKey parentKey, long id) {
+    return reader.load(entityType, parentKey, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, DatastoreKey parentKey, String id) {
-    return reader.load(entityClass, parentKey, id);
+  public <E> E load(Type entityType, DatastoreKey parentKey, String id) {
+    return reader.load(entityType, parentKey, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, DatastoreKey key) {
-    return reader.load(entityClass, key);
+  public <E> E load(Type entityType, DatastoreKey key) {
+    return reader.load(entityType, key);
   }
 
   @Override
-  public <E> List<E> loadById(Class<E> entityClass, List<Long> identifiers) {
-    return reader.loadById(entityClass, identifiers);
+  public <E> List<E> loadById(Type entityType, List<Long> identifiers) {
+    return reader.loadById(entityType, identifiers);
   }
 
   @Override
-  public <E> List<E> loadByName(Class<E> entityClass, List<String> identifiers) {
-    return reader.loadByName(entityClass, identifiers);
+  public <E> List<E> loadByKey(Type entityType, List<DatastoreKey> keys) {
+    return reader.loadByKey(entityType, keys);
   }
 
   @Override
-  public <E> List<E> loadByKey(Class<E> entityClass, List<DatastoreKey> keys) {
-    return reader.loadByKey(entityClass, keys);
+  public <E> List<E> loadByName(Type entityType, List<String> identifiers) {
+    return reader.loadByName(entityType, identifiers);
   }
 
   @Override
@@ -341,13 +404,13 @@ public KeyQueryRequest createKeyQueryRequest(String query) {
   }
 
   @Override
-  public <E> QueryResponse<E> executeEntityQueryRequest(Class<E> expectedResultType,
-      EntityQueryRequest request) {
+  public <E> QueryResponse<E> executeEntityQueryRequest(Type expectedResultType,
+                                                        EntityQueryRequest request) {
     return reader.executeEntityQueryRequest(expectedResultType, request);
   }
 
   @Override
-  public <E> QueryResponse<E> executeProjectionQueryRequest(Class<E> expectedResultType,
+  public <E> QueryResponse<E> executeProjectionQueryRequest(Type expectedResultType,
       ProjectionQueryRequest request) {
     return reader.executeProjectionQueryRequest(expectedResultType, request);
   }
diff --git a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreWriter.java b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreWriter.java
index ff5f51a..6871e89 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreWriter.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/DefaultDatastoreWriter.java
@@ -21,6 +21,7 @@
 import static com.jmethods.catatumbo.impl.DatastoreUtils.toNativeEntities;
 import static com.jmethods.catatumbo.impl.DatastoreUtils.toNativeFullEntities;
 
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -107,15 +108,16 @@ public DefaultDatastoreWriter(DefaultDatastoreTransaction transaction) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> E insert(E entity) {
+  public <E> E insert(E entity, Type entityType) {
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_INSERT, entity);
+      entityManager.executeEntityListeners(CallbackType.PRE_INSERT, entity, entityType);
       FullEntity<?> nativeEntity = (FullEntity<?>) Marshaller.marshal(entityManager, entity,
-          Intent.INSERT);
+              Intent.INSERT, entityType);
+      Type entityClass = entityType != null ? entityType : entity.getClass();
       Entity insertedNativeEntity = nativeWriter.add(nativeEntity);
       @SuppressWarnings("unchecked")
-      E insertedEntity = (E) Unmarshaller.unmarshal(insertedNativeEntity, entity.getClass());
-      entityManager.executeEntityListeners(CallbackType.POST_INSERT, insertedEntity);
+      E insertedEntity = (E) Unmarshaller.unmarshal(insertedNativeEntity, entityClass);
+      entityManager.executeEntityListeners(CallbackType.POST_INSERT, insertedEntity, entityType);
       return insertedEntity;
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -134,17 +136,18 @@ public <E> E insert(E entity) {
    *           if any error occurs while inserting.
    */
   @SuppressWarnings("unchecked")
-  public <E> List<E> insert(List<E> entities) {
+  public <E> List<E> insert(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return new ArrayList<>();
     }
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_INSERT, entities);
-      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager, Intent.INSERT);
-      Class<?> entityClass = entities.get(0).getClass();
+      entityManager.executeEntityListeners(CallbackType.PRE_INSERT, entities, entityType);
+      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager,
+              Intent.INSERT, entityType);
+      Type entityClass = entityType != null ? entityType : entities.get(0).getClass();
       List<Entity> insertedNativeEntities = nativeWriter.add(nativeEntities);
       List<E> insertedEntities = (List<E>) toEntities(entityClass, insertedNativeEntities);
-      entityManager.executeEntityListeners(CallbackType.POST_INSERT, insertedEntities);
+      entityManager.executeEntityListeners(CallbackType.POST_INSERT, insertedEntities, entityType);
       return insertedEntities;
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -162,14 +165,15 @@ public <E> List<E> insert(List<E> entities) {
    *           if any error occurs while updating.
    */
   @SuppressWarnings("unchecked")
-  public <E> E update(E entity) {
+  public <E> E update(E entity, Type entityType) {
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entity);
+      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entity, entityType);
       Intent intent = (nativeWriter instanceof Batch) ? Intent.BATCH_UPDATE : Intent.UPDATE;
-      Entity nativeEntity = (Entity) Marshaller.marshal(entityManager, entity, intent);
+      Entity nativeEntity = (Entity) Marshaller.marshal(entityManager, entity, intent, entityType);
+      Type entityClass = entityType != null ? entityType : entity.getClass();
       nativeWriter.update(nativeEntity);
-      E updatedEntity = (E) Unmarshaller.unmarshal(nativeEntity, entity.getClass());
-      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntity);
+      E updatedEntity = (E) Unmarshaller.unmarshal(nativeEntity, entityClass);
+      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntity, entityType);
       return updatedEntity;
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -188,18 +192,18 @@ public <E> E update(E entity) {
    *           if any error occurs while inserting.
    */
   @SuppressWarnings("unchecked")
-  public <E> List<E> update(List<E> entities) {
+  public <E> List<E> update(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return new ArrayList<>();
     }
     try {
-      Class<E> entityClass = (Class<E>) entities.get(0).getClass();
-      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entities);
+      Type entityClass = entityType != null ? entityType : entities.get(0).getClass();
+      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entities, entityType);
       Intent intent = (nativeWriter instanceof Batch) ? Intent.BATCH_UPDATE : Intent.UPDATE;
-      Entity[] nativeEntities = toNativeEntities(entities, entityManager, intent);
+      Entity[] nativeEntities = toNativeEntities(entities, entityManager, intent, entityType);
       nativeWriter.update(nativeEntities);
       List<E> updatedEntities = toEntities(entityClass, nativeEntities);
-      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntities);
+      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntities, entityType);
       return updatedEntities;
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -214,12 +218,12 @@ public <E> List<E> update(List<E> entities) {
    *          the entity to update
    * @return the updated entity which may be different than the given entity.
    */
-  public <E> E updateWithOptimisticLock(E entity) {
-    PropertyMetadata versionMetadata = EntityIntrospector.getVersionMetadata(entity);
+  public <E> E updateWithOptimisticLock(E entity, Type entityType) {
+    PropertyMetadata versionMetadata = EntityIntrospector.getVersionMetadata(entity, entityType);
     if (versionMetadata == null) {
-      return update(entity);
+      return update(entity, entityType);
     } else {
-      return updateWithOptimisticLockingInternal(entity, versionMetadata);
+      return updateWithOptimisticLockingInternal(entity, versionMetadata, entityType);
     }
 
   }
@@ -232,16 +236,17 @@ public <E> E updateWithOptimisticLock(E entity) {
    *          the entities to update
    * @return the updated entities
    */
-  public <E> List<E> updateWithOptimisticLock(List<E> entities) {
+  public <E> List<E> updateWithOptimisticLock(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return new ArrayList<>();
     }
-    Class<?> entityClass = entities.get(0).getClass();
+    Type entityClass = entityType != null ? entityType : entities.get(0).getClass();
+
     PropertyMetadata versionMetadata = EntityIntrospector.getVersionMetadata(entityClass);
     if (versionMetadata == null) {
-      return update(entities);
+      return update(entities, entityType);
     } else {
-      return updateWithOptimisticLockInternal(entities, versionMetadata);
+      return updateWithOptimisticLockInternal(entities, versionMetadata, entityType);
     }
   }
 
@@ -255,11 +260,14 @@ public <E> List<E> updateWithOptimisticLock(List<E> entities) {
    * @return the updated entity
    */
   @SuppressWarnings("unchecked")
-  private <E> E updateWithOptimisticLockingInternal(E entity, PropertyMetadata versionMetadata) {
+  private <E> E updateWithOptimisticLockingInternal(E entity,
+                                                    PropertyMetadata versionMetadata,
+                                                    Type entityType) {
     Transaction transaction = null;
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entity);
-      Entity nativeEntity = (Entity) Marshaller.marshal(entityManager, entity, Intent.UPDATE);
+      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entity, entityType);
+      Entity nativeEntity = (Entity) Marshaller.marshal(entityManager, entity,
+              Intent.UPDATE, entityType);
       transaction = datastore.newTransaction();
       Entity storedNativeEntity = transaction.get(nativeEntity.getKey());
       if (storedNativeEntity == null) {
@@ -275,8 +283,9 @@ private <E> E updateWithOptimisticLockingInternal(E entity, PropertyMetadata ver
       }
       transaction.update(nativeEntity);
       transaction.commit();
-      E updatedEntity = (E) Unmarshaller.unmarshal(nativeEntity, entity.getClass());
-      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntity);
+      Type entityClass = entityType != null ? entityType : entity.getClass();
+      E updatedEntity = (E) Unmarshaller.unmarshal(nativeEntity, entityClass);
+      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntity, entityType);
       return updatedEntity;
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -289,6 +298,7 @@ private <E> E updateWithOptimisticLockingInternal(E entity, PropertyMetadata ver
    * Internal worker method for updating the entities using optimistic locking.
    * 
    * @param entities
+      Entity[] nativeEntities = toNativeEntities(entities, entityManager, intent, type);
    *          the entities to update
    * @param versionMetadata
    *          the metadata of the version property
@@ -296,11 +306,13 @@ private <E> E updateWithOptimisticLockingInternal(E entity, PropertyMetadata ver
    */
   @SuppressWarnings("unchecked")
   private <E> List<E> updateWithOptimisticLockInternal(List<E> entities,
-      PropertyMetadata versionMetadata) {
+                                                       PropertyMetadata versionMetadata,
+                                                       Type entityType) {
     Transaction transaction = null;
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entities);
-      Entity[] nativeEntities = toNativeEntities(entities, entityManager, Intent.UPDATE);
+      entityManager.executeEntityListeners(CallbackType.PRE_UPDATE, entities, entityType);
+      Entity[] nativeEntities = toNativeEntities(entities, entityManager,
+              Intent.UPDATE, entityType);
       // The above native entities already have the version incremented by
       // the marshalling process
       Key[] nativeKeys = new Key[nativeEntities.length];
@@ -326,8 +338,9 @@ private <E> List<E> updateWithOptimisticLockInternal(List<E> entities,
       }
       transaction.update(nativeEntities);
       transaction.commit();
-      List<E> updatedEntities = (List<E>) toEntities(entities.get(0).getClass(), nativeEntities);
-      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntities);
+      Type entityClass = entityType != null ? entityType : entities.get(0).getClass();
+      List<E> updatedEntities = (List<E>) toEntities(entityClass, nativeEntities);
+      entityManager.executeEntityListeners(CallbackType.POST_UPDATE, updatedEntities, entityType);
       return updatedEntities;
 
     } catch (DatastoreException exp) {
@@ -348,15 +361,17 @@ private <E> List<E> updateWithOptimisticLockInternal(List<E> entities,
    * @throws EntityManagerException
    *           if any error occurs while saving.
    */
-  public <E> E upsert(E entity) {
+  public <E> E upsert(E entity, Type entityType) {
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_UPSERT, entity);
+      entityManager.executeEntityListeners(CallbackType.PRE_UPSERT, entity, entityType);
       FullEntity<?> nativeEntity = (FullEntity<?>) Marshaller.marshal(entityManager, entity,
-          Intent.UPSERT);
+              Intent.UPSERT, entityType);
       Entity upsertedNativeEntity = nativeWriter.put(nativeEntity);
       @SuppressWarnings("unchecked")
-      E upsertedEntity = (E) Unmarshaller.unmarshal(upsertedNativeEntity, entity.getClass());
-      entityManager.executeEntityListeners(CallbackType.POST_UPSERT, upsertedEntity);
+      Type entityClass = entityType != null ? entityType : entity.getClass();
+
+      E upsertedEntity = (E) Unmarshaller.unmarshal(upsertedNativeEntity, entityClass);
+      entityManager.executeEntityListeners(CallbackType.POST_UPSERT, upsertedEntity, entityType);
       return upsertedEntity;
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -374,17 +389,18 @@ public <E> E upsert(E entity) {
    *           if any error occurs while saving.
    */
   @SuppressWarnings("unchecked")
-  public <E> List<E> upsert(List<E> entities) {
+  public <E> List<E> upsert(List<E> entities, Type entityType) {
     if (entities == null || entities.isEmpty()) {
       return new ArrayList<>();
     }
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_UPSERT, entities);
-      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager, Intent.UPSERT);
-      Class<?> entityClass = entities.get(0).getClass();
+      entityManager.executeEntityListeners(CallbackType.PRE_UPSERT, entities, entityType);
+      FullEntity<?>[] nativeEntities = toNativeFullEntities(entities, entityManager,
+              Intent.UPSERT, entityType);
+      Type entityClass = entityType != null ? entityType : entities.get(0).getClass();
       List<Entity> upsertedNativeEntities = nativeWriter.put(nativeEntities);
       List<E> upsertedEntities = (List<E>) toEntities(entityClass, upsertedNativeEntities);
-      entityManager.executeEntityListeners(CallbackType.POST_UPSERT, upsertedEntities);
+      entityManager.executeEntityListeners(CallbackType.POST_UPSERT, upsertedEntities, entityType);
       return upsertedEntities;
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
@@ -399,12 +415,12 @@ public <E> List<E> upsert(List<E> entities) {
    * @throws EntityManagerException
    *           if any error occurs while deleting.
    */
-  public void delete(Object entity) {
+  public void delete(Object entity, Type entityType) {
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_DELETE, entity);
-      Key nativeKey = Marshaller.marshalKey(entityManager, entity);
+      entityManager.executeEntityListeners(CallbackType.PRE_DELETE, entity, entityType);
+      Key nativeKey = Marshaller.marshalKey(entityManager, entity, entityType);
       nativeWriter.delete(nativeKey);
-      entityManager.executeEntityListeners(CallbackType.POST_DELETE, entity);
+      entityManager.executeEntityListeners(CallbackType.POST_DELETE, entity, entityType);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
     }
@@ -418,15 +434,15 @@ public void delete(Object entity) {
    * @throws EntityManagerException
    *           if any error occurs while deleting.
    */
-  public void delete(List<?> entities) {
+  public void delete(List<?> entities, Type entityType) {
     try {
-      entityManager.executeEntityListeners(CallbackType.PRE_DELETE, entities);
+      entityManager.executeEntityListeners(CallbackType.PRE_DELETE, entities, entityType);
       Key[] nativeKeys = new Key[entities.size()];
       for (int i = 0; i < entities.size(); i++) {
-        nativeKeys[i] = Marshaller.marshalKey(entityManager, entities.get(i));
+        nativeKeys[i] = Marshaller.marshalKey(entityManager, entities.get(i), entityType);
       }
       nativeWriter.delete(nativeKeys);
-      entityManager.executeEntityListeners(CallbackType.POST_DELETE, entities);
+      entityManager.executeEntityListeners(CallbackType.POST_DELETE, entities, entityType);
     } catch (DatastoreException exp) {
       throw DatastoreUtils.wrap(exp);
     }
@@ -436,16 +452,16 @@ public void delete(List<?> entities) {
    * Deletes the entity with the given ID. The entity is assumed to be a root entity (no parent).
    * The entity kind will be determined from the supplied entity class.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity.
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> void delete(Class<E> entityClass, long id) {
+  public <E> void delete(Type entityType, long id) {
     try {
-      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
       Key nativeKey = entityManager.newNativeKeyFactory().setKind(entityMetadata.getKind())
           .newKey(id);
       nativeWriter.delete(nativeKey);
@@ -458,16 +474,16 @@ public <E> void delete(Class<E> entityClass, long id) {
    * Deletes the entity with the given ID. The entity is assumed to be a root entity (no parent).
    * The entity kind will be determined from the supplied entity class.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param id
    *          the ID of the entity.
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> void delete(Class<E> entityClass, String id) {
+  public <E> void delete(Type entityType, String id) {
     try {
-      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
       Key nativeKey = entityManager.newNativeKeyFactory().setKind(entityMetadata.getKind())
           .newKey(id);
       nativeWriter.delete(nativeKey);
@@ -479,8 +495,8 @@ public <E> void delete(Class<E> entityClass, String id) {
   /**
    * Deletes the entity with the given ID and parent key.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key
    * @param id
@@ -488,9 +504,9 @@ public <E> void delete(Class<E> entityClass, String id) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id) {
+  public <E> void delete(Type entityType, DatastoreKey parentKey, long id) {
     try {
-      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
       Key nativeKey = Key.newBuilder(parentKey.nativeKey(), entityMetadata.getKind(), id).build();
       nativeWriter.delete(nativeKey);
     } catch (DatastoreException exp) {
@@ -501,8 +517,8 @@ public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id) {
   /**
    * Deletes the entity with the given ID and parent key.
    * 
-   * @param entityClass
-   *          the entity class.
+   * @param entityType
+   *          the entity type.
    * @param parentKey
    *          the parent key
    * @param id
@@ -510,9 +526,9 @@ public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id) {
    * @throws EntityManagerException
    *           if any error occurs while inserting.
    */
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, String id) {
+  public <E> void delete(Type entityType, DatastoreKey parentKey, String id) {
     try {
-      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+      EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
       Key nativeKey = Key.newBuilder(parentKey.nativeKey(), entityMetadata.getKind(), id).build();
       nativeWriter.delete(nativeKey);
     } catch (DatastoreException exp) {
diff --git a/src/main/java/com/jmethods/catatumbo/impl/DefaultEntityManager.java b/src/main/java/com/jmethods/catatumbo/impl/DefaultEntityManager.java
index 0241f16..6d256b7 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/DefaultEntityManager.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/DefaultEntityManager.java
@@ -17,6 +17,7 @@
 package com.jmethods.catatumbo.impl;
 
 import java.lang.reflect.Method;
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.EnumMap;
@@ -104,8 +105,8 @@ public Datastore getDatastore() {
   }
 
   @Override
-  public <E> long deleteAll(Class<E> entityClass) {
-    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityClass);
+  public <E> long deleteAll(Type entityType) {
+    EntityMetadata entityMetadata = EntityIntrospector.introspect(entityType);
     return deleteAll(entityMetadata.getKind());
   }
 
@@ -172,62 +173,102 @@ public <T> T executeInTransaction(TransactionalTask<T> task) {
 
   @Override
   public <E> E insert(E entity) {
-    return writer.insert(entity);
+    return insert(entity, null);
+  }
+
+  @Override
+  public <E> E insert(E entity, Type entityType) {
+    return writer.insert(entity, entityType);
   }
 
   @Override
   public <E> List<E> insert(List<E> entities) {
-    return writer.insert(entities);
+    return insert(entities, null);
+  }
+
+  @Override
+  public <E> List<E> insert(List<E> entities, Type entityType) {
+    return writer.insert(entities, entityType);
   }
 
   @Override
   public <E> E update(E entity) {
-    return writer.updateWithOptimisticLock(entity);
+    return update(entity, null);
+  }
+
+  @Override
+  public <E> E update(E entity, Type entityType) {
+    return writer.updateWithOptimisticLock(entity, entityType);
   }
 
   @Override
   public <E> List<E> update(List<E> entities) {
-    return writer.updateWithOptimisticLock(entities);
+    return update(entities, null);
+  }
+
+  @Override
+  public <E> List<E> update(List<E> entities, Type entityType) {
+    return writer.updateWithOptimisticLock(entities, entityType);
   }
 
   @Override
   public <E> E upsert(E entity) {
-    return writer.upsert(entity);
+    return upsert(entity, null);
+  }
+
+  @Override
+  public <E> E upsert(E entity, Type entityType) {
+    return writer.upsert(entity, entityType);
   }
 
   @Override
   public <E> List<E> upsert(List<E> entities) {
-    return writer.upsert(entities);
+    return upsert(entities, null);
+  }
+
+  @Override
+  public <E> List<E> upsert(List<E> entities, Type entityType) {
+    return writer.upsert(entities, entityType);
   }
 
   @Override
   public void delete(Object entity) {
-    writer.delete(entity);
+    delete(entity, null);
+  }
+
+  @Override
+  public void delete(Object entity, Type entityType) {
+    writer.delete(entity, entityType);
   }
 
   @Override
   public void delete(List<?> entities) {
-    writer.delete(entities);
+    delete(entities, null);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, long id) {
-    writer.delete(entityClass, id);
+  public void delete(List<?> entities, Type entityType) {
+    writer.delete(entities, entityType);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, String id) {
-    writer.delete(entityClass, id);
+  public void delete(Type entityType, long id) {
+    writer.delete(entityType, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, long id) {
-    writer.delete(entityClass, parentKey, id);
+  public void delete(Type entityType, String id) {
+    writer.delete(entityType, id);
   }
 
   @Override
-  public <E> void delete(Class<E> entityClass, DatastoreKey parentKey, String id) {
-    writer.delete(entityClass, parentKey, id);
+  public void delete(Type entityType, DatastoreKey parentKey, long id) {
+    writer.delete(entityType, parentKey, id);
+  }
+
+  @Override
+  public void delete(Type entityType, DatastoreKey parentKey, String id) {
+    writer.delete(entityType, parentKey, id);
   }
 
   @Override
@@ -241,43 +282,43 @@ public void deleteByKey(List<DatastoreKey> keys) {
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, long id) {
-    return reader.load(entityClass, id);
+  public <E> E load(Type entityType, long id) {
+    return reader.load(entityType, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, String id) {
-    return reader.load(entityClass, id);
+  public <E> E load(Type entityType, String id) {
+    return reader.load(entityType, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, DatastoreKey parentKey, long id) {
-    return reader.load(entityClass, parentKey, id);
+  public <E> E load(Type entityType, DatastoreKey parentKey, long id) {
+    return reader.load(entityType, parentKey, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, DatastoreKey parentKey, String id) {
-    return reader.load(entityClass, parentKey, id);
+  public <E> E load(Type entityType, DatastoreKey parentKey, String id) {
+    return reader.load(entityType, parentKey, id);
   }
 
   @Override
-  public <E> E load(Class<E> entityClass, DatastoreKey key) {
-    return reader.load(entityClass, key);
+  public <E> E load(Type entityType, DatastoreKey key) {
+    return reader.load(entityType, key);
   }
 
   @Override
-  public <E> List<E> loadById(Class<E> entityClass, List<Long> identifiers) {
-    return reader.loadById(entityClass, identifiers);
+  public <E> List<E> loadById(Type entityType, List<Long> identifiers) {
+    return reader.loadById(entityType, identifiers);
   }
 
   @Override
-  public <E> List<E> loadByName(Class<E> entityClass, List<String> identifiers) {
-    return reader.loadByName(entityClass, identifiers);
+  public <E> List<E> loadByKey(Type entityType, List<DatastoreKey> keys) {
+    return reader.loadByKey(entityType, keys);
   }
 
   @Override
-  public <E> List<E> loadByKey(Class<E> entityClass, List<DatastoreKey> keys) {
-    return reader.loadByKey(entityClass, keys);
+  public <E> List<E> loadByName(Type entityType, List<String> identifiers) {
+    return reader.loadByName(entityType, identifiers);
   }
 
   @Override
@@ -296,13 +337,13 @@ public KeyQueryRequest createKeyQueryRequest(String query) {
   }
 
   @Override
-  public <E> QueryResponse<E> executeEntityQueryRequest(Class<E> expectedResultType,
-      EntityQueryRequest request) {
+  public <E> QueryResponse<E> executeEntityQueryRequest(Type expectedResultType,
+                                                        EntityQueryRequest request) {
     return reader.executeEntityQueryRequest(expectedResultType, request);
   }
 
   @Override
-  public <E> QueryResponse<E> executeProjectionQueryRequest(Class<E> expectedResultType,
+  public <E> QueryResponse<E> executeProjectionQueryRequest(Type expectedResultType,
       ProjectionQueryRequest request) {
     return reader.executeProjectionQueryRequest(expectedResultType, request);
   }
@@ -324,14 +365,25 @@ public DatastoreStats getDatastoreStats() {
 
   @Override
   public DatastoreKey allocateId(Object entity) {
-    List<DatastoreKey> keys = allocateId(Arrays.asList(new Object[] { entity }));
+    return allocateId(entity, null);
+  }
+
+  @Override
+  public DatastoreKey allocateId(Object entity, Type entityType) {
+    List<DatastoreKey> keys = allocateId(Arrays.asList(new Object[] { entity }), entityType);
     return keys.get(0);
   }
 
   @Override
   public List<DatastoreKey> allocateId(List<Object> entities) {
+    return allocateId(entities, null);
+  }
+
+  @Override
+  public List<DatastoreKey> allocateId(List<Object> entities, Type entityType) {
     for (Object entity : entities) {
-      IdentifierMetadata identifierMetadata = EntityIntrospector.getIdentifierMetadata(entity);
+      IdentifierMetadata identifierMetadata = EntityIntrospector
+              .getIdentifierMetadata(entity, entityType);
       if (DataType.STRING == identifierMetadata.getDataType()) {
         throw new IllegalArgumentException(
             "ID allocation is only valid for entities with numeric identifiers");
@@ -345,7 +397,7 @@ public List<DatastoreKey> allocateId(List<Object> entities) {
     IncompleteKey[] incompleteKeys = new IncompleteKey[entities.size()];
     int i = 0;
     for (Object entity : entities) {
-      incompleteKeys[i++] = getIncompleteKey(entity);
+      incompleteKeys[i++] = getIncompleteKey(entity, entityType);
     }
     List<Key> nativeKeys = datastore.allocateId(incompleteKeys);
     return DatastoreUtils.toDatastoreKeys(nativeKeys);
@@ -358,8 +410,8 @@ public List<DatastoreKey> allocateId(List<Object> entities) {
    *          the entity
    * @return the incomplete key
    */
-  private IncompleteKey getIncompleteKey(Object entity) {
-    EntityMetadata entityMetadata = EntityIntrospector.introspect(entity.getClass());
+  private IncompleteKey getIncompleteKey(Object entity, Type entityType) {
+    EntityMetadata entityMetadata = EntityIntrospector.introspect(entity, entityType);
     String kind = entityMetadata.getKind();
     ParentKeyMetadata parentKeyMetadata = entityMetadata.getParentKeyMetadata();
     DatastoreKey parentKey = null;
@@ -420,14 +472,14 @@ private void putDefaultCallback(CallbackType callbackType, CallbackMetadata meta
    * @param entity
    *          the entity that produced the event
    */
-  public void executeEntityListeners(CallbackType callbackType, Object entity) {
+  public void executeEntityListeners(CallbackType callbackType, Object entity, Type entityType) {
     // We may get null entities here. For example loading a nonexistent ID
     // or IDs.
     if (entity == null) {
       return;
     }
     EntityListenersMetadata entityListenersMetadata = EntityIntrospector
-        .getEntityListenersMetadata(entity);
+            .getEntityListenersMetadata(entity, entityType);
     List<CallbackMetadata> callbacks = entityListenersMetadata.getCallbacks(callbackType);
     if (!entityListenersMetadata.isExcludeDefaultListeners()) {
       executeGlobalListeners(callbackType, entity);
@@ -460,9 +512,9 @@ public void executeEntityListeners(CallbackType callbackType, Object entity) {
    * @param entities
    *          the entities
    */
-  public void executeEntityListeners(CallbackType callbackType, List<?> entities) {
+  public void executeEntityListeners(CallbackType callbackType, List<?> entities, Type entityType) {
     for (Object entity : entities) {
-      executeEntityListeners(callbackType, entity);
+      executeEntityListeners(callbackType, entity, entityType);
     }
   }
 
diff --git a/src/main/java/com/jmethods/catatumbo/impl/EmbeddableIntrospector.java b/src/main/java/com/jmethods/catatumbo/impl/EmbeddableIntrospector.java
index e2f9262..5cfe3c1 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/EmbeddableIntrospector.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/EmbeddableIntrospector.java
@@ -99,7 +99,8 @@ private void processFields() {
    *          the field to process
    */
   private void processSimpleField(Field field) {
-    PropertyMetadata propertyMetadata = IntrospectionUtils.getPropertyMetadata(field);
+    PropertyMetadata propertyMetadata = IntrospectionUtils
+            .getPropertyMetadata(field, field.getType());
     if (propertyMetadata != null) {
       metadata.putPropertyMetadata(propertyMetadata);
     }
@@ -122,7 +123,8 @@ private void processEmbeddedField(Field field) {
     }
     boolean indexed = embeddedAnnotation.indexed();
     boolean optional = embeddedAnnotation.optional();
-    PropertyMetadata propertyMetadata = new PropertyMetadata(field, name, indexed, optional);
+    PropertyMetadata propertyMetadata = new PropertyMetadata(field, field.getType(), name,
+            indexed, optional);
     metadata.putPropertyMetadata(propertyMetadata);
   }
 
diff --git a/src/main/java/com/jmethods/catatumbo/impl/EmbeddedField.java b/src/main/java/com/jmethods/catatumbo/impl/EmbeddedField.java
index 6c13a8a..fde1179 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/EmbeddedField.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/EmbeddedField.java
@@ -32,6 +32,11 @@ public class EmbeddedField {
    */
   private Field field;
 
+  /**
+   * Type of the underlying field
+   */
+  private Class<?> type;
+
   /**
    * Parent embedded field, if any
    */
@@ -51,7 +56,19 @@ public class EmbeddedField {
    *          the underlying field
    */
   public EmbeddedField(Field field) {
-    this(field, null);
+    this(field, null, field.getType());
+  }
+
+  /**
+   * Creates a new instance of <code>EmbeddedField</code>.
+   * 
+   * @param field
+   *            the underlying field
+   * @param type
+   *         the type of the field if need to be specified
+   */
+  public EmbeddedField(Field field, Class<?> type) {
+    this(field, null, type);
   }
 
   /**
@@ -63,14 +80,29 @@ public EmbeddedField(Field field) {
    *          the parent embedded field. May be <code>null</code>.
    */
   public EmbeddedField(Field field, EmbeddedField parent) {
+    this(field, parent, field.getType());
+  }
+
+  /**
+   * Creates a new instance of <code>EmbeddedField</code>.
+   *
+   * @param field
+   *            the underlying field
+   * @param parent
+   *            the parent embedded field. May be <code>null</code>.
+   * @param type
+   *         the type of the field if need to be specified
+   */
+  public EmbeddedField(Field field, EmbeddedField parent, Class<?> type) {
     this.field = field;
     this.parent = parent;
+    this.type = type;
     computeQualifiedName();
   }
 
   /**
    * Returns the underlying field.
-   * 
+   *
    * @return the field the underlying field.
    */
   public Field getField() {
@@ -101,7 +133,7 @@ public String getQualifiedName() {
    * @return the type of this field.
    */
   public Class<?> getType() {
-    return field.getType();
+    return type;
   }
 
   /**
diff --git a/src/main/java/com/jmethods/catatumbo/impl/EmbeddedIntrospector.java b/src/main/java/com/jmethods/catatumbo/impl/EmbeddedIntrospector.java
index 948d4d7..4495950 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/EmbeddedIntrospector.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/EmbeddedIntrospector.java
@@ -135,7 +135,8 @@ private void processFields() {
    *          the field to process
    */
   private void processSimpleField(Field child) {
-    PropertyMetadata propertyMetadata = IntrospectionUtils.getPropertyMetadata(child);
+    PropertyMetadata propertyMetadata = IntrospectionUtils
+            .getPropertyMetadata(child, child.getType());
     if (propertyMetadata != null) {
       // Process override
       processPropertyOverride(propertyMetadata);
diff --git a/src/main/java/com/jmethods/catatumbo/impl/EmbeddedMetadata.java b/src/main/java/com/jmethods/catatumbo/impl/EmbeddedMetadata.java
index 17dd14b..2c640c6 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/EmbeddedMetadata.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/EmbeddedMetadata.java
@@ -72,8 +72,8 @@ public EmbeddedMetadata(EmbeddedField field) {
     this.field = field;
     // Default storage strategy is EXPLODED
     this.storageStrategy = StorageStrategy.EXPLODED;
-    this.readMethod = IntrospectionUtils.findReadMethodHandle(field.getField());
-    this.writeMethod = IntrospectionUtils.findWriteMethodHandle(field.getField());
+    this.readMethod = IntrospectionUtils.findReadMethodHandle(field.getField(), field.getType());
+    this.writeMethod = IntrospectionUtils.findWriteMethodHandle(field.getField(), field.getType());
   }
 
   /**
diff --git a/src/main/java/com/jmethods/catatumbo/impl/EntityIntrospector.java b/src/main/java/com/jmethods/catatumbo/impl/EntityIntrospector.java
index 29edec6..46c591f 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/EntityIntrospector.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/EntityIntrospector.java
@@ -17,6 +17,9 @@
 package com.jmethods.catatumbo.impl;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.lang.reflect.TypeVariable;
 import java.time.OffsetDateTime;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
@@ -70,12 +73,18 @@ public class EntityIntrospector {
   /**
    * Cache of introspected classes
    */
-  private static final Cache<Class<?>, EntityMetadata> cache = new Cache<>(64);
+  private static final Cache<Type, EntityMetadata> cache = new Cache<>(64);
 
   /**
-   * The class to introspect
+   * The type to introspect
    */
-  private final Class<?> entityClass;
+  private final Type entityType;
+
+
+  /**
+   * The raw class corresponding to the type
+   */
+  private final Class<?> rawType;
 
   /**
    * Output of the introspection, the metadata about the entity
@@ -85,11 +94,12 @@ public class EntityIntrospector {
   /**
    * Creates a new instance of <code>EntityIntrospector</code>.
    *
-   * @param entityClass
-   *          the entity class to introspect
+   * @param entityType
+   *          the entity type to introspect
    */
-  private EntityIntrospector(Class<?> entityClass) {
-    this.entityClass = entityClass;
+  private EntityIntrospector(Type entityType) {
+    this.entityType = entityType;
+    this.rawType = resolveRawType(entityType);
   }
 
   /**
@@ -99,40 +109,40 @@ private EntityIntrospector(Class<?> entityClass) {
    *          the entity object to introspect.
    * @return the metadata of the entity
    */
-  public static EntityMetadata introspect(Object entity) {
-    return introspect(entity.getClass());
+  public static EntityMetadata introspect(Object entity, Type entityType) {
+    return introspect(entityType != null ? entityType : entity.getClass());
   }
 
   /**
    * Introspects the given entity class and returns the metadata of the entity.
    *
-   * @param entityClass
-   *          the entity class to introspect
+   * @param entityType
+   *            the entity type to introspect
    * @return the metadata of the entity
    */
-  public static EntityMetadata introspect(Class<?> entityClass) {
-    EntityMetadata cachedMetadata = cache.get(entityClass);
+  public static EntityMetadata introspect(Type entityType) {
+    EntityMetadata cachedMetadata = cache.get(entityType);
     if (cachedMetadata != null) {
       return cachedMetadata;
     }
-    return loadMetadata(entityClass);
+    return loadMetadata(entityType);
   }
 
   /**
    * Loads the metadata for the given class and puts it in the cache and returns it.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @return the metadata for the given class
    */
-  private static EntityMetadata loadMetadata(Class<?> entityClass) {
-    synchronized (entityClass) {
-      EntityMetadata metadata = cache.get(entityClass);
+  private static EntityMetadata loadMetadata(Type entityType) {
+    synchronized (entityType) {
+      EntityMetadata metadata = cache.get(entityType);
       if (metadata == null) {
-        EntityIntrospector introspector = new EntityIntrospector(entityClass);
+        EntityIntrospector introspector = new EntityIntrospector(entityType);
         introspector.process();
         metadata = introspector.entityMetadata;
-        cache.put(entityClass, metadata);
+        cache.put(entityType, metadata);
       }
       return metadata;
     }
@@ -142,32 +152,73 @@ private static EntityMetadata loadMetadata(Class<?> entityClass) {
    * Processes the entity class using reflection and builds the metadata.
    */
   private void process() {
-    Entity entity = entityClass.getAnnotation(Entity.class);
-    ProjectedEntity projectedEntity = entityClass.getAnnotation(ProjectedEntity.class);
+    Entity entity = rawType.getAnnotation(Entity.class);
+    ProjectedEntity projectedEntity = rawType.getAnnotation(ProjectedEntity.class);
     if (entity != null) {
       initEntityMetadata(entity);
     } else if (projectedEntity != null) {
       initEntityMetadata(projectedEntity);
     } else {
       String message = String.format("Class %s must either have %s or %s annotation",
-          entityClass.getName(), Entity.class.getName(), ProjectedEntity.class.getName());
+              rawType.getName(), Entity.class.getName(), ProjectedEntity.class.getName());
       throw new EntityManagerException(message);
     }
 
+    processGenericParameters();
     processPropertyOverrides();
 
     processFields();
     // If we did not find valid Identifier...
     if (entityMetadata.getIdentifierMetadata() == null) {
       throw new EntityManagerException(
-          String.format("Class %s requires a field with annotation of %s", entityClass.getName(),
+          String.format("Class %s requires a field with annotation of %s", rawType.getName(),
               Identifier.class.getName()));
     }
-    entityMetadata.setEntityListenersMetadata(EntityListenersIntrospector.introspect(entityClass));
+    entityMetadata.setEntityListenersMetadata(EntityListenersIntrospector.introspect(rawType));
     entityMetadata.ensureUniqueProperties();
     entityMetadata.cleanup();
   }
 
+  /**
+   * Resolve the raw class from a given type
+   *
+   * @param type
+   *          the type to resolve
+   * @return the raw class of the underlying type
+   */
+  private Class<?> resolveRawType(Type type) {
+    if (type instanceof Class) {
+      return (Class<?>) type;
+    } else if (type instanceof ParameterizedType) {
+      return (Class<?>) ((ParameterizedType) type).getRawType();
+    } else {
+      String message = String.format("Given type %s not supported."
+              + "Only Class and ParameterizedType are valid.", type.getTypeName());
+      throw new EntityManagerException(message);
+    }
+  }
+
+  /**
+   * Resolve the field type if generic
+   *
+   * @param field
+   *          the field to resolve
+   * @return the actual class of the field
+   */
+  private Class<?> reifyFieldType(Field field) {
+    // Resolve the field type if it's a generic parameterizedType
+    Class<?> fieldType = field.getType();
+    if (field.getGenericType() instanceof TypeVariable) {
+      Class<?> reifiedType = entityMetadata.reify((TypeVariable) field.getGenericType());
+      if (reifiedType != null) {
+        fieldType = reifiedType;
+      }
+    }
+
+    return fieldType;
+  }
+
+
   /**
    * Initializes the metadata using the given {@link Entity} annotation.
    * 
@@ -177,9 +228,9 @@ private void process() {
   private void initEntityMetadata(Entity entity) {
     String kind = entity.kind();
     if (kind.trim().length() == 0) {
-      kind = entityClass.getSimpleName();
+      kind = rawType.getSimpleName();
     }
-    entityMetadata = new EntityMetadata(entityClass, kind);
+    entityMetadata = new EntityMetadata(rawType, kind);
   }
 
   /**
@@ -191,17 +242,36 @@ private void initEntityMetadata(Entity entity) {
   private void initEntityMetadata(ProjectedEntity projectedEntity) {
     String kind = projectedEntity.kind();
     if (kind.trim().length() == 0) {
-      String message = String.format("Class %s requires a non-blank Kind", entityClass.getName());
+      String message = String.format("Class %s requires a non-blank Kind", rawType.getName());
       throw new EntityManagerException(message);
     }
-    entityMetadata = new EntityMetadata(entityClass, kind, true);
+    entityMetadata = new EntityMetadata(rawType, kind, true);
+  }
+
+  /**
+   * Resolve generic parameters if a {@link ParameterizedType} was given.
+   */
+  private void processGenericParameters() {
+    if (entityType instanceof ParameterizedType) {
+      TypeVariable<? extends Class<?>>[] typeParameters = rawType.getTypeParameters();
+      Type[] reified = ((ParameterizedType) entityType).getActualTypeArguments();
+      if (typeParameters.length != reified.length) {
+        String message = String.format("ParameterizedType %s requires %d generic types",
+                entityType.getTypeName(), typeParameters.length);
+        throw new EntityManagerException(message);
+      }
+
+      for (int i = 0; i < typeParameters.length; i++) {
+        entityMetadata.putGenericType(typeParameters[i], reified[i]);
+      }
+    }
   }
 
   /**
    * Processes the property overrides for the embedded objects, if any.
    */
   private void processPropertyOverrides() {
-    PropertyOverrides propertyOverrides = entityClass.getAnnotation(PropertyOverrides.class);
+    PropertyOverrides propertyOverrides = rawType.getAnnotation(PropertyOverrides.class);
     if (propertyOverrides == null) {
       return;
     }
@@ -217,16 +287,17 @@ private void processPropertyOverrides() {
   private void processFields() {
     List<Field> fields = getAllFields();
     for (Field field : fields) {
+      Class<?> fieldType = reifyFieldType(field);
       if (field.isAnnotationPresent(Identifier.class)) {
-        processIdentifierField(field);
+        processIdentifierField(field, fieldType);
       } else if (field.isAnnotationPresent(Key.class)) {
-        processKeyField(field);
+        processKeyField(field, fieldType);
       } else if (field.isAnnotationPresent(ParentKey.class)) {
-        processParentKeyField(field);
+        processParentKeyField(field, fieldType);
       } else if (field.isAnnotationPresent(Embedded.class)) {
-        processEmbeddedField(field);
+        processEmbeddedField(field, fieldType);
       } else {
-        processField(field);
+        processField(field, fieldType);
       }
     }
   }
@@ -239,7 +310,7 @@ private void processFields() {
    */
   private List<Field> getAllFields() {
     List<Field> allFields = new ArrayList<>();
-    Class<?> clazz = entityClass;
+    Class<?> clazz = rawType;
     boolean stop;
     do {
       List<Field> fields = IntrospectionUtils.getPersistableFields(clazz);
@@ -255,11 +326,13 @@ private List<Field> getAllFields() {
    *
    * @param field
    *          the identifier field
+   * @param type
+   *          the field Type
    */
-  private void processIdentifierField(Field field) {
+  private void processIdentifierField(Field field, Class<?> type) {
     Identifier identifier = field.getAnnotation(Identifier.class);
     boolean autoGenerated = identifier.autoGenerated();
-    IdentifierMetadata identifierMetadata = new IdentifierMetadata(field, autoGenerated);
+    IdentifierMetadata identifierMetadata = new IdentifierMetadata(field, type, autoGenerated);
     entityMetadata.setIdentifierMetadata(identifierMetadata);
   }
 
@@ -268,16 +341,17 @@ private void processIdentifierField(Field field) {
    * 
    * @param field
    *          the Key field
+   * @param type
+   *          the field Type
    */
-  private void processKeyField(Field field) {
+  private void processKeyField(Field field, Class<?> type) {
     String fieldName = field.getName();
-    Class<?> type = field.getType();
     if (!type.equals(DatastoreKey.class)) {
       String message = String.format("Invalid type, %s, for Key field %s in class %s. ", type,
-          fieldName, entityClass);
+              fieldName, rawType);
       throw new EntityManagerException(message);
     }
-    KeyMetadata keyMetadata = new KeyMetadata(field);
+    KeyMetadata keyMetadata = new KeyMetadata(field, type);
     entityMetadata.setKeyMetadata(keyMetadata);
   }
 
@@ -286,16 +360,17 @@ private void processKeyField(Field field) {
    * 
    * @param field
    *          the ParentKey field
+   * @param type
+   *          the field type
    */
-  private void processParentKeyField(Field field) {
+  private void processParentKeyField(Field field, Class<?> type) {
     String fieldName = field.getName();
-    Class<?> type = field.getType();
     if (!type.equals(DatastoreKey.class)) {
       String message = String.format("Invalid type, %s, for ParentKey field %s in class %s. ", type,
-          fieldName, entityClass);
+              fieldName, rawType);
       throw new EntityManagerException(message);
     }
-    ParentKeyMetadata parentKeyMetadata = new ParentKeyMetadata(field);
+    ParentKeyMetadata parentKeyMetadata = new ParentKeyMetadata(field, type);
     entityMetadata.setParentKetMetadata(parentKeyMetadata);
   }
 
@@ -304,13 +379,15 @@ private void processParentKeyField(Field field) {
    *
    * @param field
    *          the field to process
+   * @param type
+   *          the field type
    */
-  private void processField(Field field) {
-    PropertyMetadata propertyMetadata = IntrospectionUtils.getPropertyMetadata(field);
+  private void processField(Field field, Class<?> type) {
+    PropertyMetadata propertyMetadata = IntrospectionUtils.getPropertyMetadata(field, type);
     if (propertyMetadata != null) {
       // If the field is from a super class, there might be some
       // overrides, so process those.
-      if (!field.getDeclaringClass().equals(entityClass)) {
+      if (!field.getDeclaringClass().equals(rawType)) {
         applyPropertyOverride(propertyMetadata);
       }
       entityMetadata.putPropertyMetadata(propertyMetadata);
@@ -335,7 +412,7 @@ private void processVersionField(PropertyMetadata propertyMetadata) {
     if (!long.class.equals(dataClass)) {
       String messageFormat = "Field %s in class %s must be of type %s";
       throw new EntityManagerException(String.format(messageFormat,
-          propertyMetadata.getField().getName(), entityClass, long.class));
+          propertyMetadata.getField().getName(), rawType, long.class));
     }
     entityMetadata.setVersionMetadata(propertyMetadata);
   }
@@ -373,7 +450,7 @@ private void validateAutoTimestampField(PropertyMetadata propertyMetadata) {
     if (Collections.binarySearch(VALID_TIMESTAMP_TYPES, dataClass.getName()) < 0) {
       String messageFormat = "Field %s in class %s must be one of the following types - %s";
       throw new EntityManagerException(String.format(messageFormat,
-          propertyMetadata.getField().getName(), entityClass, VALID_TIMESTAMP_TYPES));
+          propertyMetadata.getField().getName(), rawType, VALID_TIMESTAMP_TYPES));
     }
   }
 
@@ -402,11 +479,13 @@ private void applyPropertyOverride(PropertyMetadata propertyMetadata) {
    * 
    * @param field
    *          the embedded field
+   * @param type
+   *          the field type
    */
-  private void processEmbeddedField(Field field) {
-    // First create EmbeddedField so we can maintain the path/depth of the
+  private void processEmbeddedField(Field field, Class<?> type) {
+    // First, create EmbeddedField so we can maintain the path/depth of the
     // embedded field
-    EmbeddedField embeddedField = new EmbeddedField(field);
+    EmbeddedField embeddedField = new EmbeddedField(field, type);
     // Introspect the embedded field.
     EmbeddedMetadata embeddedMetadata = EmbeddedIntrospector.introspect(embeddedField,
         entityMetadata);
@@ -416,13 +495,13 @@ private void processEmbeddedField(Field field) {
   /**
    * Convenient method for getting the metadata of the field used for optimistic locking.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @return the metadata of the field used for optimistic locking. Returns <code>null</code>, if
    *         the class does not have a field with {@link Version} annotation.
    */
-  public static PropertyMetadata getVersionMetadata(Class<?> entityClass) {
-    return introspect(entityClass).getVersionMetadata();
+  public static PropertyMetadata getVersionMetadata(Type entityType) {
+    return introspect(entityType).getVersionMetadata();
   }
 
   /**
@@ -433,8 +512,8 @@ public static PropertyMetadata getVersionMetadata(Class<?> entityClass) {
    * @return the metadata of the field used for optimistic locking. Returns <code>null</code>, if
    *         the class does not have a field with {@link Version} annotation.
    */
-  public static PropertyMetadata getVersionMetadata(Object entity) {
-    return introspect(entity.getClass()).getVersionMetadata();
+  public static PropertyMetadata getVersionMetadata(Object entity, Type entityType) {
+    return introspect(entity, entityType).getVersionMetadata();
   }
 
   /**
@@ -444,19 +523,19 @@ public static PropertyMetadata getVersionMetadata(Object entity) {
    *          the entity
    * @return the Identifier Metadata
    */
-  public static IdentifierMetadata getIdentifierMetadata(Object entity) {
-    return getIdentifierMetadata(entity.getClass());
+  public static IdentifierMetadata getIdentifierMetadata(Object entity, Type entityType) {
+    return introspect(entity, entityType).getIdentifierMetadata();
   }
 
   /**
    * Returns the Identifier Metadata for the given entity class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @return the Identifier Metadata
    */
-  public static IdentifierMetadata getIdentifierMetadata(Class<?> entityClass) {
-    return introspect(entityClass).getIdentifierMetadata();
+  public static IdentifierMetadata getIdentifierMetadata(Type entityType) {
+    return introspect(entityType).getIdentifierMetadata();
 
   }
 
@@ -467,19 +546,19 @@ public static IdentifierMetadata getIdentifierMetadata(Class<?> entityClass) {
    *          the entity
    * @return the metadata of EntityListeners associated with the given entity.
    */
-  public static EntityListenersMetadata getEntityListenersMetadata(Object entity) {
-    return introspect(entity.getClass()).getEntityListenersMetadata();
+  public static EntityListenersMetadata getEntityListenersMetadata(Object entity, Type entityType) {
+    return introspect(entity, entityType).getEntityListenersMetadata();
   }
 
   /**
    * Returns the metadata of entity listeners associated with the given entity class.
    * 
-   * @param entityClass
-   *          the entity class
+   * @param entityType
+   *          the entity type
    * @return the metadata of entity listeners associated with the given entity class.
    */
-  public static EntityListenersMetadata getEntityListenersMetadata(Class<?> entityClass) {
-    return introspect(entityClass).getEntityListenersMetadata();
+  public static EntityListenersMetadata getEntityListenersMetadata(Type entityType) {
+    return introspect(entityType).getEntityListenersMetadata();
   }
 
 }
diff --git a/src/main/java/com/jmethods/catatumbo/impl/EntityMetadata.java b/src/main/java/com/jmethods/catatumbo/impl/EntityMetadata.java
index cf634ad..dd0ed31 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/EntityMetadata.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/EntityMetadata.java
@@ -17,7 +17,10 @@
 package com.jmethods.catatumbo.impl;
 
 import java.lang.annotation.Annotation;
+import java.lang.reflect.Type;
+import java.lang.reflect.TypeVariable;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 
 import com.jmethods.catatumbo.CreatedTimestamp;
@@ -103,6 +106,11 @@ public class EntityMetadata extends MetadataBase {
    */
   private EntityListenersMetadata entityListenersMetadata;
 
+  /**
+   * Mapping of potential type parameters and their concrete type
+   */
+  private Map<TypeVariable<?>, Type> genericTypes;
+
   /**
    * Creates a new instance of <code>EntityMetadata</code>.
    *
@@ -132,6 +140,7 @@ public EntityMetadata(Class<?> entityClass, String kind, boolean projectedEntity
     this.projectedEntity = projectedEntity;
     propertyOverrideMap = new HashMap<>();
     masterPropertyMetadataMap = new HashMap<>();
+    genericTypes = new LinkedHashMap<>();
   }
 
   /**
@@ -359,6 +368,14 @@ public void updateMasterPropertyMetadataMap(String mappedName, String qualifiedN
 
   }
 
+  public Class<?> reify(TypeVariable<?> typeParameter) {
+    return (Class<?>) genericTypes.get(typeParameter);
+  }
+
+  public void putGenericType(TypeVariable<?> typeParameter, Type concreteType) {
+    genericTypes.put(typeParameter, concreteType);
+  }
+
   /**
    * Returns the metadata of the entity listeners.
    * 
diff --git a/src/main/java/com/jmethods/catatumbo/impl/FieldMetadata.java b/src/main/java/com/jmethods/catatumbo/impl/FieldMetadata.java
index a0099bf..7c51a00 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/FieldMetadata.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/FieldMetadata.java
@@ -32,6 +32,11 @@ public abstract class FieldMetadata {
    */
   protected final Field field;
 
+  /**
+   * Underlying field type
+   */
+  protected final Class<?> type;
+
   /**
    * Read method (or getter method) for this field
    */
@@ -47,12 +52,15 @@ public abstract class FieldMetadata {
    * 
    * @param field
    *          the field
+   * @param type
+   *          the field Type
    *
    */
-  public FieldMetadata(Field field) {
+  public FieldMetadata(Field field, Class<?> type) {
     this.field = field;
-    this.readMethod = IntrospectionUtils.findReadMethodHandle(this.field);
-    this.writeMethod = IntrospectionUtils.findWriteMethodHandle(this.field);
+    this.type = type;
+    this.readMethod = IntrospectionUtils.findReadMethodHandle(this.field, this.type);
+    this.writeMethod = IntrospectionUtils.findWriteMethodHandle(this.field, this.type);
   }
 
   /**
@@ -98,7 +106,7 @@ public MethodHandle getWriteMethod() {
    */
   @SuppressWarnings("rawtypes")
   public Class getDeclaredType() {
-    return field.getType();
+    return type;
   }
 
 }
diff --git a/src/main/java/com/jmethods/catatumbo/impl/IdentifierMetadata.java b/src/main/java/com/jmethods/catatumbo/impl/IdentifierMetadata.java
index 06076f9..d113f2e 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/IdentifierMetadata.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/IdentifierMetadata.java
@@ -99,13 +99,15 @@ public static DataType forClass(Class<?> dataClass) {
    * 
    * @param field
    *          the field
+   * @param type
+   *          the field type
    * @param autoGenerated
    *          if the identifier is to be generated automatically
    */
-  public IdentifierMetadata(Field field, boolean autoGenerated) {
-    super(field);
+  public IdentifierMetadata(Field field, Class<?> type, boolean autoGenerated) {
+    super(field, type);
     this.autoGenerated = autoGenerated;
-    DataType dataType = DataType.forClass(field.getType());
+    DataType dataType = DataType.forClass(type);
     if (dataType == null) {
       idClassMetadata = new IdClassMetadata(getDeclaredType());
       dataType = DataType.forClass(idClassMetadata.getIdType());
@@ -114,7 +116,7 @@ public IdentifierMetadata(Field field, boolean autoGenerated) {
     }
     if (dataType == null) {
       String message = String.format("Invalid identifier type %s for field %s in class %s",
-          field.getType().getName(), field.getName(), field.getDeclaringClass().getName());
+              type.getName(), field.getName(), field.getDeclaringClass().getName());
       throw new EntityManagerException(message);
     }
     this.dataType = dataType;
diff --git a/src/main/java/com/jmethods/catatumbo/impl/IntrospectionUtils.java b/src/main/java/com/jmethods/catatumbo/impl/IntrospectionUtils.java
index 0da232b..cdfbd1a 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/IntrospectionUtils.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/IntrospectionUtils.java
@@ -77,15 +77,17 @@ public static Object instantiate(MetadataBase metadata) {
    * 
    * @param field
    *          the field whose metadata has to be prepared
+   * @param type
+   *          the field Type
    * @return metadata of the given field.
    */
-  public static PropertyMetadata getPropertyMetadata(Field field) {
+  public static PropertyMetadata getPropertyMetadata(Field field, Class<?> type) {
     Property property = field.getAnnotation(Property.class);
     // For fields that have @Property annotation, we expect both setter and
     // getter methods. For all other fields, we only treat them as
     // persistable if we find valid getter and setter methods.
     try {
-      PropertyMetadata propertyMetadata = new PropertyMetadata(field);
+      PropertyMetadata propertyMetadata = new PropertyMetadata(field, type);
       return propertyMetadata;
     } catch (NoAccessorMethodException | NoMutatorMethodException exp) {
       if (property != null) {
@@ -101,15 +103,17 @@ public static PropertyMetadata getPropertyMetadata(Field field) {
    * 
    * @param field
    *          the field
-   * 
+   * @param type
+   *          the field Type
+   *
    * @return the {@link MethodHandle} for reading the field's value
    * @throws EntityManagerException
    *           if no read method exists
    */
-  public static MethodHandle findReadMethodHandle(Field field) {
+  public static MethodHandle findReadMethodHandle(Field field, Class<?> type) {
     String readMethodName;
     MethodHandle mh = null;
-    if (boolean.class.equals(field.getType())) {
+    if (boolean.class.equals(type)) {
       readMethodName = IntrospectionUtils.getReadMethodNameForBoolean(field);
       mh = findInstanceMethod(field.getDeclaringClass(), readMethodName, field.getType());
     }
@@ -131,12 +135,14 @@ public static MethodHandle findReadMethodHandle(Field field) {
    * 
    * @param field
    *          the field
-   * 
+   * @param type
+   *          the field Type
+   *
    * @return the {@link MethodHandle} to update the field
    * @throws NoMutatorMethodException
    *           if no matching method exists
    */
-  public static MethodHandle findWriteMethodHandle(Field field) {
+  public static MethodHandle findWriteMethodHandle(Field field, Class<?> type) {
     ConstructorMetadata constructorMetadata = ConstructorIntrospector
         .introspect(field.getDeclaringClass());
     Class<?> containerClass;
@@ -145,7 +151,7 @@ public static MethodHandle findWriteMethodHandle(Field field) {
       containerClass = constructorMetadata.getBuilderClass();
       for (String prefix : WRITE_METHOD_PREFIXES) {
         mh = findInstanceMethod(containerClass, getWriteMethodName(field, prefix), null,
-            field.getType());
+                field.getType());
         if (mh != null) {
           break;
         }
diff --git a/src/main/java/com/jmethods/catatumbo/impl/KeyMetadata.java b/src/main/java/com/jmethods/catatumbo/impl/KeyMetadata.java
index 96353f8..dc16f82 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/KeyMetadata.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/KeyMetadata.java
@@ -30,9 +30,11 @@ public class KeyMetadata extends FieldMetadata {
    * 
    * @param field
    *          the field.
+   * @param type
+   *          the field Type.
    */
-  public KeyMetadata(Field field) {
-    super(field);
+  public KeyMetadata(Field field, Class<?> type) {
+    super(field, type);
   }
 
 }
diff --git a/src/main/java/com/jmethods/catatumbo/impl/Marshaller.java b/src/main/java/com/jmethods/catatumbo/impl/Marshaller.java
index 3d17a60..9bef241 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/Marshaller.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/Marshaller.java
@@ -16,6 +16,7 @@
 
 package com.jmethods.catatumbo.impl;
 
+import java.lang.reflect.Type;
 import java.time.Instant;
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
@@ -166,11 +167,12 @@ public boolean isValidOnProjectedEntities() {
    * @param intent
    *          the intent of marshalling
    */
-  private Marshaller(DefaultEntityManager entityManager, Object entity, Intent intent) {
+  private Marshaller(DefaultEntityManager entityManager, Object entity, Intent intent,
+                     Type entityType) {
     this.entityManager = entityManager;
     this.entity = entity;
     this.intent = intent;
-    entityMetadata = EntityIntrospector.introspect(entity.getClass());
+    entityMetadata = EntityIntrospector.introspect(entity, entityType);
     validateIntent();
   }
 
@@ -200,12 +202,35 @@ private void validateIntent() {
    *          the purpose. For example, if the purpose if INSERT or UPSERT, the marshaller would
    *          auto generate any keys. Where as if the purpose is UPDATE, then then marshaller will
    *          NOT generate any keys.
+   * @param entityType
+   *          the entity type
+   * @return the marshaled object
+   */
+  @SuppressWarnings("rawtypes")
+  public static BaseEntity marshal(DefaultEntityManager entityManager, Object entity, Intent intent,
+                                   Type entityType) {
+    Marshaller marshaller = new Marshaller(entityManager, entity, intent, entityType);
+    return marshaller.marshal();
+  }
+
+  /**
+   * Marshals the given entity (POJO) into the format needed for the low level Cloud Datastore API.
+   *
+   * @param entityManager
+   *          the entity manager
+   * @param entity
+   *          the entity to marshal
+   * @param intent
+   *          the intent or purpose of marshalling. Marshalling process varies slightly depending on
+   *          the purpose. For example, if the purpose if INSERT or UPSERT, the marshaller would
+   *          auto generate any keys. Where as if the purpose is UPDATE, then then marshaller will
+   *          NOT generate any keys.
    * @return the marshaled object
    */
   @SuppressWarnings("rawtypes")
   public static BaseEntity marshal(DefaultEntityManager entityManager, Object entity,
-      Intent intent) {
-    Marshaller marshaller = new Marshaller(entityManager, entity, intent);
+                                   Intent intent) {
+    Marshaller marshaller = new Marshaller(entityManager, entity, intent, null);
     return marshaller.marshal();
   }
 
@@ -240,10 +265,12 @@ private BaseEntity<?> marshal() {
    *          the entity manager.
    * @param entity
    *          the entity from which key is to be extracted
+   * @param entityType
+   *          the entity type
    * @return extracted key.
    */
-  public static Key marshalKey(DefaultEntityManager entityManager, Object entity) {
-    Marshaller marshaller = new Marshaller(entityManager, entity, Intent.DELETE);
+  public static Key marshalKey(DefaultEntityManager entityManager, Object entity, Type entityType) {
+    Marshaller marshaller = new Marshaller(entityManager, entity, Intent.DELETE, entityType);
     marshaller.marshalKey();
     return (Key) marshaller.key;
   }
diff --git a/src/main/java/com/jmethods/catatumbo/impl/ParentKeyMetadata.java b/src/main/java/com/jmethods/catatumbo/impl/ParentKeyMetadata.java
index 130b371..e819925 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/ParentKeyMetadata.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/ParentKeyMetadata.java
@@ -30,9 +30,11 @@ public class ParentKeyMetadata extends KeyMetadata {
    * 
    * @param field
    *          the field
+   * @param type
+   *          the field type
    */
-  public ParentKeyMetadata(Field field) {
-    super(field);
+  public ParentKeyMetadata(Field field, Class<?> type) {
+    super(field, type);
   }
 
 }
diff --git a/src/main/java/com/jmethods/catatumbo/impl/PropertyMetadata.java b/src/main/java/com/jmethods/catatumbo/impl/PropertyMetadata.java
index ab8d980..3ac8c72 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/PropertyMetadata.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/PropertyMetadata.java
@@ -78,9 +78,11 @@ public class PropertyMetadata extends FieldMetadata {
    *
    * @param field
    *          the field
+   * @param type
+   *          the field type
    */
-  public PropertyMetadata(Field field) {
-    super(field);
+  public PropertyMetadata(Field field, Class<?> type) {
+    super(field, type);
     String mappedName = field.getName();
     boolean indexed = true;
     boolean optional = false;
@@ -105,6 +107,8 @@ public PropertyMetadata(Field field) {
    * 
    * @param field
    *          the field
+   * @param type
+   *          the field type
    * @param mappedName
    *          name of the property in the Datastore
    * @param indexed
@@ -112,8 +116,9 @@ public PropertyMetadata(Field field) {
    * @param optional
    *          whether or not the property is optional
    */
-  public PropertyMetadata(Field field, String mappedName, boolean indexed, boolean optional) {
-    super(field);
+  public PropertyMetadata(Field field, Class<?> type, String mappedName, boolean indexed,
+                          boolean optional) {
+    super(field, type);
     this.mappedName = mappedName;
     this.indexed = indexed;
     setOptional(optional);
@@ -195,7 +200,7 @@ public boolean isOptional() {
    *          whether or not the field represented by this metadata is optional.
    */
   public void setOptional(boolean optional) {
-    if (field.getType().isPrimitive() || field.isAnnotationPresent(Version.class)
+    if (getDeclaredType().isPrimitive() || field.isAnnotationPresent(Version.class)
         || field.isAnnotationPresent(CreatedTimestamp.class)
         || field.isAnnotationPresent(UpdatedTimestamp.class)) {
       this.optional = false;
@@ -244,7 +249,7 @@ public Mapper getMapper() {
    */
   private Mapper initializeMapper() {
     try {
-      return MapperFactory.getInstance().getMapper(field);
+      return MapperFactory.getInstance().getMapper(field, getDeclaredType());
     } catch (NoSuitableMapperException exp) {
       String message = String.format(
           "No suitable mapper found or error occurred creating a mapper for field %s in class %s",
diff --git a/src/main/java/com/jmethods/catatumbo/impl/Unmarshaller.java b/src/main/java/com/jmethods/catatumbo/impl/Unmarshaller.java
index 5243690..f3c1d44 100644
--- a/src/main/java/com/jmethods/catatumbo/impl/Unmarshaller.java
+++ b/src/main/java/com/jmethods/catatumbo/impl/Unmarshaller.java
@@ -17,6 +17,7 @@
 package com.jmethods.catatumbo.impl;
 
 import java.lang.invoke.MethodHandle;
+import java.lang.reflect.Type;
 import java.util.Collection;
 
 import com.google.cloud.datastore.BaseEntity;
@@ -57,12 +58,12 @@ public class Unmarshaller {
    * 
    * @param nativeEntity
    *          the native entity to unmarshal
-   * @param entityClass
+   * @param entityType
    *          the expected model type
    */
-  private Unmarshaller(BaseEntity<?> nativeEntity, Class<?> entityClass) {
+  private Unmarshaller(BaseEntity<?> nativeEntity, Type entityType) {
     this.nativeEntity = nativeEntity;
-    entityMetadata = EntityIntrospector.introspect(entityClass);
+    entityMetadata = EntityIntrospector.introspect(entityType);
 
   }
 
@@ -73,13 +74,13 @@ private Unmarshaller(BaseEntity<?> nativeEntity, Class<?> entityClass) {
    *          target object type
    * @param nativeEntity
    *          the native Entity
-   * @param entityClass
+   * @param entityType
    *          the target type
    * @return Object that is equivalent to the given native entity. If the given
    *         <code>datastoreEntity</code> is <code>null</code>, returns <code>null</code>.
    */
-  public static <T> T unmarshal(Entity nativeEntity, Class<T> entityClass) {
-    return unmarshalBaseEntity(nativeEntity, entityClass);
+  public static <T> T unmarshal(Entity nativeEntity, Type entityType) {
+    return unmarshalBaseEntity(nativeEntity, entityType);
   }
 
   /**
@@ -89,13 +90,13 @@ public static <T> T unmarshal(Entity nativeEntity, Class<T> entityClass) {
    *          target object type
    * @param nativeEntity
    *          the native Entity
-   * @param entityClass
+   * @param entityType
    *          the target type
    * @return Object that is equivalent to the given native entity. If the given
    *         <code>datastoreEntity</code> is <code>null</code>, returns <code>null</code>.
    */
-  public static <T> T unmarshal(ProjectionEntity nativeEntity, Class<T> entityClass) {
-    return unmarshalBaseEntity(nativeEntity, entityClass);
+  public static <T> T unmarshal(ProjectionEntity nativeEntity, Type entityType) {
+    return unmarshalBaseEntity(nativeEntity, entityType);
   }
 
   /**
@@ -133,15 +134,15 @@ private <T> T unmarshal() {
    * 
    * @param nativeEntity
    *          the native entity to unmarshal
-   * @param entityClass
+   * @param entityType
    *          the target type of the model class
    * @return the model object
    */
-  private static <T> T unmarshalBaseEntity(BaseEntity<?> nativeEntity, Class<T> entityClass) {
+  private static <T> T unmarshalBaseEntity(BaseEntity<?> nativeEntity, Type entityType) {
     if (nativeEntity == null) {
       return null;
     }
-    Unmarshaller unmarshaller = new Unmarshaller(nativeEntity, entityClass);
+    Unmarshaller unmarshaller = new Unmarshaller(nativeEntity, entityType);
     return unmarshaller.unmarshal();
   }
 
diff --git a/src/test/java/com/jmethods/catatumbo/entities/GenericEntity.java b/src/test/java/com/jmethods/catatumbo/entities/GenericEntity.java
new file mode 100644
index 0000000..270380f
--- /dev/null
+++ b/src/test/java/com/jmethods/catatumbo/entities/GenericEntity.java
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2016 Sai Pullabhotla.
+ *
+ * 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.jmethods.catatumbo.entities;
+
+import com.jmethods.catatumbo.Embedded;
+import com.jmethods.catatumbo.Entity;
+import com.jmethods.catatumbo.Identifier;
+import com.jmethods.catatumbo.PropertyMapper;
+import com.jmethods.catatumbo.custommappers.CurrencyMapper;
+
+import java.math.BigDecimal;
+import java.util.Objects;
+
+/**
+ * @author Aurelien Thieriot
+ *
+ */
+@Entity
+public class GenericEntity<T, Z> {
+
+  @Identifier
+  private long id;
+
+  @Embedded
+  private T embeddedGeneric;
+
+  private Z generic;
+
+  /**
+   * @return the id
+   */
+  public long getId() {
+    return id;
+  }
+
+  /**
+   * @param id
+   *          the id to set
+   */
+  public void setId(long id) {
+    this.id = id;
+  }
+
+  public T getEmbeddedGeneric() {
+    return embeddedGeneric;
+  }
+
+  public void setEmbeddedGeneric(T embeddedGeneric) {
+    this.embeddedGeneric = embeddedGeneric;
+  }
+
+  public Z getGeneric() {
+    return generic;
+  }
+
+  public void setGeneric(Z generic) {
+    this.generic = generic;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    GenericEntity<?, ?> that = (GenericEntity<?, ?>) o;
+    return id == that.id &&
+            Objects.equals(embeddedGeneric, that.embeddedGeneric) &&
+            Objects.equals(generic, that.generic);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(id, embeddedGeneric, generic);
+  }
+
+  public static GenericEntity<Address, String> createSampleGenericEntity1() {
+    GenericEntity<Address, String> genericEntity = new GenericEntity<>();
+    genericEntity.setEmbeddedGeneric(Address.getSample1());
+    genericEntity.setGeneric("Stuff");
+    return genericEntity;
+  }
+}
diff --git a/src/test/java/com/jmethods/catatumbo/entities/GenericParameterizedType.java b/src/test/java/com/jmethods/catatumbo/entities/GenericParameterizedType.java
new file mode 100644
index 0000000..9465d63
--- /dev/null
+++ b/src/test/java/com/jmethods/catatumbo/entities/GenericParameterizedType.java
@@ -0,0 +1,29 @@
+package com.jmethods.catatumbo.entities;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+
+public class GenericParameterizedType implements ParameterizedType {
+
+  private Type[] types;
+
+  public GenericParameterizedType(Type[] types) {
+    this.types = types;
+  }
+
+  @Override
+  public Type[] getActualTypeArguments() {
+    return types;
+  }
+
+  @Override
+  public Type getRawType() {
+    return GenericEntity.class;
+  }
+
+  @Override
+  public Type getOwnerType() {
+    return null;
+  }
+}
diff --git a/src/test/java/com/jmethods/catatumbo/impl/EntityIntrospectorTest.java b/src/test/java/com/jmethods/catatumbo/impl/EntityIntrospectorTest.java
index 0fa0988..f9c348d 100644
--- a/src/test/java/com/jmethods/catatumbo/impl/EntityIntrospectorTest.java
+++ b/src/test/java/com/jmethods/catatumbo/impl/EntityIntrospectorTest.java
@@ -23,9 +23,13 @@
 import static org.junit.Assert.assertTrue;
 
 import java.awt.Button;
+import java.lang.reflect.Type;
 import java.util.Map;
 import java.util.Objects;
 
+import com.jmethods.catatumbo.entities.Address;
+import com.jmethods.catatumbo.entities.GenericEntity;
+import com.jmethods.catatumbo.entities.GenericParameterizedType;
 import org.junit.Test;
 
 import com.jmethods.catatumbo.EntityManagerException;
@@ -66,6 +70,29 @@
  */
 public class EntityIntrospectorTest {
 
+  @Test
+  public void testIntrospect_Generic() {
+    Type[] types = {Address.class, String.class};
+    EntityMetadata metadata = EntityIntrospector.introspect(GenericEntity.class, new GenericParameterizedType(types));
+    System.out.println(metadata);
+    System.out.println("************");
+    metadata = EntityIntrospector.introspect(GenericEntity.class, new GenericParameterizedType(types));
+    System.out.println(metadata);
+
+    assertEquals(metadata.reify(GenericEntity.class.getTypeParameters()[0]), types[0]);
+    assertEquals(metadata.reify(GenericEntity.class.getTypeParameters()[1]), types[1]);
+  }
+
+  @Test(expected = EntityManagerException.class)
+  public void testIntrospect_invalid_Generic_Types() {
+    EntityIntrospector.introspect(GenericEntity.class, new GenericParameterizedType(new Type[] {String.class}));
+  }
+
+  @Test(expected = EntityManagerException.class)
+  public void testIntrospect_Generic_Type_missing() {
+    EntityIntrospector.introspect(GenericEntity.class);
+  }
+
   @Test
   public void testIntrospect_Embedded() {
     EntityMetadata metadata = EntityIntrospector.introspect(Customer.class);
diff --git a/src/test/java/com/jmethods/catatumbo/impl/MarshallerTest.java b/src/test/java/com/jmethods/catatumbo/impl/MarshallerTest.java
index 3de4983..6a2691d 100644
--- a/src/test/java/com/jmethods/catatumbo/impl/MarshallerTest.java
+++ b/src/test/java/com/jmethods/catatumbo/impl/MarshallerTest.java
@@ -21,6 +21,9 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+import com.jmethods.catatumbo.entities.Address;
+import com.jmethods.catatumbo.entities.GenericEntity;
+import com.jmethods.catatumbo.entities.GenericParameterizedType;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
@@ -36,6 +39,8 @@
 import com.jmethods.catatumbo.entities.WrappedStringIdEntity;
 import com.jmethods.catatumbo.impl.Marshaller.Intent;
 
+import java.lang.reflect.Type;
+
 /**
  * @author Sai Pullabhotla
  *
@@ -49,6 +54,15 @@ public static void setUpBeforeClass() throws Exception {
     em = (DefaultEntityManager) TestUtils.getEntityManager();
   }
 
+  @Test
+  public void testMarshal_Generic() {
+    Type[] types = {Address.class, String.class};
+    GenericEntity<Address, String> genericEntity = GenericEntity.createSampleGenericEntity1();
+    FullEntity<?> entity = (FullEntity<?>) Marshaller.marshal(em, genericEntity, Intent.INSERT, new GenericParameterizedType(types));
+    assertEquals("Stuff", entity.getString("generic"));
+    assertEquals("Omaha", entity.getString("city"));
+  }
+
   @Test
   public void testMarshal_Embedded() {
     Customer customer = Customer.createSampleCustomer2();