gora-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lewi...@apache.org
Subject [10/37] gora git commit: Refactored the code
Date Wed, 23 Aug 2017 20:55:08 GMT
http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
index 113d0c4..1d9093d 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
@@ -17,35 +17,8 @@
 
 package org.apache.gora.cassandra.store;
 
-import com.datastax.driver.core.Cluster;
-import com.datastax.driver.core.ConsistencyLevel;
-import com.datastax.driver.core.HostDistance;
-import com.datastax.driver.core.KeyspaceMetadata;
-import com.datastax.driver.core.PoolingOptions;
-import com.datastax.driver.core.ProtocolOptions;
-import com.datastax.driver.core.ProtocolVersion;
-import com.datastax.driver.core.QueryOptions;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.SocketOptions;
-import com.datastax.driver.core.TableMetadata;
-import com.datastax.driver.core.TypeCodec;
-import com.datastax.driver.core.policies.ConstantReconnectionPolicy;
-import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
-import com.datastax.driver.core.policies.DefaultRetryPolicy;
-import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
-import com.datastax.driver.core.policies.ExponentialReconnectionPolicy;
-import com.datastax.driver.core.policies.FallthroughRetryPolicy;
-import com.datastax.driver.core.policies.LatencyAwarePolicy;
-import com.datastax.driver.core.policies.LoggingRetryPolicy;
-import com.datastax.driver.core.policies.RoundRobinPolicy;
-import com.datastax.driver.core.policies.TokenAwarePolicy;
-import com.datastax.driver.mapping.Mapper;
-import com.datastax.driver.mapping.MappingManager;
-import org.apache.gora.cassandra.bean.CassandraKey;
-import org.apache.gora.cassandra.bean.ClusterKeyField;
-import org.apache.gora.cassandra.bean.Field;
-import org.apache.gora.cassandra.bean.KeySpace;
-import org.apache.gora.cassandra.bean.PartitionKeyField;
+import org.apache.gora.cassandra.persistent.CassandraNativePersistent;
+import org.apache.gora.cassandra.serializers.CassandraSerializer;
 import org.apache.gora.persistency.BeanFactory;
 import org.apache.gora.persistency.Persistent;
 import org.apache.gora.query.PartitionQuery;
@@ -53,18 +26,11 @@ import org.apache.gora.query.Query;
 import org.apache.gora.query.Result;
 import org.apache.gora.store.DataStore;
 import org.apache.gora.store.DataStoreFactory;
-import org.jdom.Attribute;
-import org.jdom.Document;
-import org.jdom.Element;
-import org.jdom.JDOMException;
-import org.jdom.input.SAXBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.List;
-import java.util.Locale;
 import java.util.Properties;
 
 /**
@@ -81,19 +47,23 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
 
   private BeanFactory<K, T> beanFactory;
 
-  private Cluster cluster;
+  private Class<K> keyClass;
 
-  private Class keyClass;
-
-  private Class persistentClass;
+  private Class<T> persistentClass;
 
   private CassandraMapping mapping;
 
-  private boolean isUseNativeSerialization;
+  private CassandraSerializer cassandraSerializer;
+
+  public enum SerializerType {
+    AVRO("AVRO"), NATIVE("NATIVE");
+    String val;
 
-  private Mapper<T> mapper;
+    SerializerType(String v) {
+      this.val = v;
+    }
+  }
 
-  private Session session;
 
   public CassandraStore() {
     super();
@@ -113,22 +83,11 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
       this.keyClass = keyClass;
       this.persistentClass = persistentClass;
       String mappingFile = DataStoreFactory.getMappingFile(properties, this, DEFAULT_MAPPING_FILE);
-      List<String> codecs = readCustomCodec(properties);
-      mapping = readMapping(mappingFile);
-      isUseNativeSerialization = Boolean.parseBoolean(properties.getProperty(CassandraStoreParameters.USE_CASSANDRA_NATIVE_SERIALIZATION));
-      Cluster.Builder builder = Cluster.builder();
-      builder = populateSettings(builder, properties);
-      this.cluster = builder.build();
-      if (codecs != null) {
-        registerCustomCodecs(codecs);
-      }
-      this.session = this.cluster.connect();
-      if (isUseNativeSerialization) {
-        this.createSchema();
-        MappingManager mappingManager = new MappingManager(session);
-        mapper = mappingManager.mapper(persistentClass);
-      }
-
+      CassandraMappingBuilder mappingBuilder = new CassandraMappingBuilder(this);
+      mapping = mappingBuilder.readMapping(mappingFile);
+      CassandraClient cassandraClient = new CassandraClient();
+      cassandraClient.initialize(properties);
+      cassandraSerializer = CassandraSerializer.getSerializer(cassandraClient, properties.getProperty(CassandraStoreParameters.CASSANDRA_SERIALIZATION_TYPE),
keyClass, persistentClass, mapping);
     } catch (Exception e) {
       LOG.error("Error while initializing Cassandra store: {}",
               new Object[]{e.getMessage()});
@@ -136,501 +95,13 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
     }
   }
 
-  private void registerCustomCodecs(List<String> codecs) throws Exception {
-    for (String codec : codecs) {
-      this.cluster.getConfiguration().getCodecRegistry().register((TypeCodec<?>) Class.forName(codec).newInstance());
-    }
-  }
-
-  /**
-   * In this method we reads the mapping file and creates the Cassandra Mapping.
-   *
-   * @param filename mapping file name
-   * @return @{@link CassandraMapping}
-   * @throws IOException
-   */
-  private CassandraMapping readMapping(String filename) throws IOException {
-    CassandraMapping map = new CassandraMapping();
-    try {
-      SAXBuilder builder = new SAXBuilder();
-      Document doc = builder.build(getClass().getClassLoader().getResourceAsStream(filename));
-
-      List<Element> keyspaces = doc.getRootElement().getChildren("keyspace");
-      List<Element> classes = doc.getRootElement().getChildren("class");
-      List<Element> keys = doc.getRootElement().getChildren("cassandraKey");
-
-      boolean classMatched = false;
-      for (Element classElement : classes) {
-        if (classElement.getAttributeValue("keyClass").equals(
-                keyClass.getCanonicalName())
-                && classElement.getAttributeValue("name").equals(
-                persistentClass.getCanonicalName())) {
-
-          classMatched = true;
-          String tableName = classElement.getAttributeValue("table");
-          map.setCoreName(tableName);
-
-          List classAttributes = classElement.getAttributes();
-          for (Object anAttributeList : classAttributes) {
-            Attribute attribute = (Attribute) anAttributeList;
-            String attributeName = attribute.getName();
-            String attributeValue = attribute.getValue();
-            map.addProperty(attributeName, attributeValue);
-          }
-
-          List<Element> fields = classElement.getChildren("field");
-
-          for (Element field : fields) {
-            Field cassandraField = new Field();
-
-            List fieldAttributes = field.getAttributes();
-            processAttributes(fieldAttributes, cassandraField);
-            map.addCassandraField(cassandraField);
-          }
-          break;
-        }
-        LOG.warn("Check that 'keyClass' and 'name' parameters in gora-solr-mapping.xml "
-                + "match with intended values. A mapping mismatch has been found therefore
"
-                + "no mapping has been initialized for class mapping at position "
-                + " {} in mapping file.", classes.indexOf(classElement));
-      }
-      if (!classMatched) {
-        LOG.error("Check that 'keyClass' and 'name' parameters in {} no mapping has been
initialized for {} class mapping", filename, persistentClass);
-      }
-
-      String keyspaceName = map.getProperty("keyspace");
-      if (keyspaceName != null) {
-        KeySpace keyspace;
-        for (Element keyspaceElement : keyspaces) {
-          if (keyspaceName.equals(keyspaceElement.getAttributeValue("name"))) {
-            keyspace = new KeySpace();
-            List fieldAttributes = keyspaceElement.getAttributes();
-            for (Object attributeObject : fieldAttributes) {
-              Attribute attribute = (Attribute) attributeObject;
-              String attributeName = attribute.getName();
-              String attributeValue = attribute.getValue();
-              switch (attributeName) {
-                case "name":
-                  keyspace.setName(attributeValue);
-                  break;
-                case "durableWrite":
-                  keyspace.setDurableWritesEnabled(Boolean.parseBoolean(attributeValue));
-                  break;
-                default:
-                  keyspace.addProperty(attributeName, attributeValue);
-                  break;
-              }
-            }
-            Element placementStrategy = keyspaceElement.getChild("placementStrategy");
-            switch (KeySpace.PlacementStrategy.valueOf(placementStrategy.getAttributeValue("name")))
{
-              case SimpleStrategy:
-                keyspace.setPlacementStrategy(KeySpace.PlacementStrategy.SimpleStrategy);
-                keyspace.setReplicationFactor(Integer.parseInt(placementStrategy.getAttributeValue("replication_factor")));
-                break;
-              case NetworkTopologyStrategy:
-                List<Element> dataCenters = placementStrategy.getChildren("datacenter");
-                keyspace.setPlacementStrategy(KeySpace.PlacementStrategy.NetworkTopologyStrategy);
-                for (Element dataCenter : dataCenters) {
-                  String dataCenterName = dataCenter.getAttributeValue("name");
-                  Integer dataCenterReplicationFactor = Integer.valueOf(dataCenter.getAttributeValue("replication_factor"));
-                  keyspace.addDataCenter(dataCenterName, dataCenterReplicationFactor);
-                }
-                break;
-            }
-            map.setKeySpace(keyspace);
-            break;
-          }
-
-        }
-
-      }
-
-      for (Element key : keys) {
-        if (keyClass.getName().equals(key.getAttributeValue("name"))) {
-          CassandraKey cassandraKey = new CassandraKey(keyClass.getName());
-          Element partitionKeys = key.getChild("partitionKey");
-          Element clusterKeys = key.getChild("clusterKey");
-          List<Element> partitionKeyFields = partitionKeys.getChildren("field");
-          List<Element> partitionCompositeKeyFields = partitionKeys.getChildren("compositeKey");
-          // process non composite partition keys
-          for (Element partitionKeyField : partitionKeyFields) {
-            PartitionKeyField fieldKey = new PartitionKeyField();
-            List fieldAttributes = partitionKeyField.getAttributes();
-            processAttributes(fieldAttributes, fieldKey);
-            cassandraKey.addPartitionKeyField(fieldKey);
-          }
-          // process composite partitions keys
-          for (Element partitionCompositeKeyField : partitionCompositeKeyFields) {
-            PartitionKeyField compositeFieldKey = new PartitionKeyField();
-            compositeFieldKey.setComposite(true);
-            List<Element> compositeKeyFields = partitionCompositeKeyField.getChildren("field");
-            for (Element partitionKeyField : compositeKeyFields) {
-              PartitionKeyField fieldKey = new PartitionKeyField();
-              List fieldAttributes = partitionKeyField.getAttributes();
-              processAttributes(fieldAttributes, fieldKey);
-              compositeFieldKey.addField(fieldKey);
-            }
-            cassandraKey.addPartitionKeyField(compositeFieldKey);
-          }
-
-          //process cluster keys
-          List<Element> clusterKeyFields = clusterKeys.getChildren("field");
-          for (Element clusterKeyField : clusterKeyFields) {
-            ClusterKeyField keyField = new ClusterKeyField();
-            List fieldAttributes = clusterKeyField.getAttributes();
-            for (Object anAttributeList : fieldAttributes) {
-              Attribute attribute = (Attribute) anAttributeList;
-              String attributeName = attribute.getName();
-              String attributeValue = attribute.getValue();
-              switch (attributeName) {
-                case "name":
-                  keyField.setFieldName(attributeValue);
-                  break;
-                case "column":
-                  keyField.setColumnName(attributeValue);
-                  break;
-                case "type":
-                  keyField.setType(attributeValue);
-                  break;
-                case "order":
-                  keyField.setOrder(ClusterKeyField.Order.valueOf(attributeValue.toUpperCase(Locale.ENGLISH)));
-                  break;
-                default:
-                  keyField.addProperty(attributeName, attributeValue);
-                  break;
-              }
-            }
-            cassandraKey.addClusterKeyField(keyField);
-          }
-          map.setCassandraKey(cassandraKey);
-        }
-      }
-    } catch (Exception ex) {
-      throw new IOException(ex);
-    }
-    return map;
-  }
-
-  private void processAttributes(List<Element> attributes, Field fieldKey) {
-    for (Object anAttributeList : attributes) {
-      Attribute attribute = (Attribute) anAttributeList;
-      String attributeName = attribute.getName();
-      String attributeValue = attribute.getValue();
-      switch (attributeName) {
-        case "name":
-          fieldKey.setFieldName(attributeValue);
-          break;
-        case "column":
-          fieldKey.setColumnName(attributeValue);
-          break;
-        case "type":
-          fieldKey.setType(attributeValue);
-          break;
-        default:
-          fieldKey.addProperty(attributeName, attributeValue);
-          break;
-      }
-    }
-  }
-
-  private List<String> readCustomCodec(Properties properties) throws JDOMException,
IOException {
-    String filename = properties.getProperty(CassandraStoreParameters.CUSTOM_CODEC_FILE);
-    if (filename != null) {
-      List<String> codecs = new ArrayList<>();
-      SAXBuilder builder = new SAXBuilder();
-      Document doc = builder.build(getClass().getClassLoader().getResourceAsStream(filename));
-      List<Element> codecElementList = doc.getRootElement().getChildren("codec");
-      for (Element codec : codecElementList) {
-        codecs.add(codec.getValue());
-      }
-      return codecs;
-    }
-    return null;
-  }
-
-  private Cluster.Builder populateSettings(Cluster.Builder builder, Properties properties)
{
-    String serversParam = properties.getProperty(CassandraStoreParameters.CASSANDRA_SERVERS);
-    String[] servers = serversParam.split(",");
-    for (String server : servers) {
-      builder = builder.addContactPoint(server);
-    }
-    String portProp = properties.getProperty(CassandraStoreParameters.PORT);
-    if (portProp != null) {
-      builder = builder.withPort(Integer.parseInt(portProp));
-    }
-    String clusterNameProp = properties.getProperty(CassandraStoreParameters.CLUSTER_NAME);
-    if (clusterNameProp != null) {
-      builder = builder.withClusterName(clusterNameProp);
-    }
-    String compressionProp = properties.getProperty(CassandraStoreParameters.COMPRESSION);
-    if (compressionProp != null) {
-      builder = builder.withCompression(ProtocolOptions.Compression.valueOf(compressionProp));
-    }
-    builder = this.populateCredentials(properties, builder);
-    builder = this.populateLoadBalancingProp(properties, builder);
-    String enableJMXProp = properties.getProperty(CassandraStoreParameters.ENABLE_JMX_REPORTING);
-    if (enableJMXProp != null) {
-      if (!Boolean.parseBoolean(enableJMXProp)) {
-        builder = builder.withoutJMXReporting();
-      }
-    }
-    String enableMetricsProp = properties.getProperty(CassandraStoreParameters.ENABLE_METRICS);
-    if (enableMetricsProp != null) {
-      if (!Boolean.parseBoolean(enableMetricsProp)) {
-        builder = builder.withoutMetrics();
-      }
-    }
-    builder = this.populatePoolingSettings(properties, builder);
-    String versionProp = properties.getProperty(CassandraStoreParameters.PROTOCOL_VERSION);
-    if (versionProp != null) {
-      builder = builder.withProtocolVersion(ProtocolVersion.fromInt(Integer.parseInt(versionProp)));
-    }
-    builder = this.populateQueryOptions(properties, builder);
-    builder = this.populateReconnectPolicy(properties, builder);
-    builder = this.populateRetrytPolicy(properties, builder);
-    builder = this.populateSocketOptions(properties, builder);
-    String enableSSLProp = properties.getProperty(CassandraStoreParameters.ENABLE_SSL);
-    if (enableSSLProp != null) {
-      if (Boolean.parseBoolean(enableSSLProp)) {
-        builder = builder.withSSL();
-      }
-    }
-    return builder;
-  }
-
-
-  private Cluster.Builder populateLoadBalancingProp(Properties properties, Cluster.Builder
builder) {
-    String loadBalancingProp = properties.getProperty(CassandraStoreParameters.LOAD_BALANCING_POLICY);
-    if (loadBalancingProp != null) {
-      switch (loadBalancingProp) {
-        case "LatencyAwareRoundRobinPolicy":
-          builder = builder.withLoadBalancingPolicy(LatencyAwarePolicy.builder(new RoundRobinPolicy()).build());
-          break;
-        case "RoundRobinPolicy":
-          builder = builder.withLoadBalancingPolicy(new RoundRobinPolicy());
-          break;
-        case "DCAwareRoundRobinPolicy": {
-          String dataCenter = properties.getProperty(CassandraStoreParameters.DATA_CENTER);
-          boolean allowRemoteDCsForLocalConsistencyLevel = Boolean.parseBoolean(
-                  properties.getProperty(CassandraStoreParameters.ALLOW_REMOTE_DCS_FOR_LOCAL_CONSISTENCY_LEVEL));
-          if (dataCenter != null && !dataCenter.isEmpty()) {
-            if (allowRemoteDCsForLocalConsistencyLevel) {
-              builder = builder.withLoadBalancingPolicy(
-                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter)
-                              .allowRemoteDCsForLocalConsistencyLevel().build());
-            } else {
-              builder = builder.withLoadBalancingPolicy(
-                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build());
-            }
-          } else {
-            if (allowRemoteDCsForLocalConsistencyLevel) {
-              builder = builder.withLoadBalancingPolicy(
-                      (DCAwareRoundRobinPolicy.builder().allowRemoteDCsForLocalConsistencyLevel().build()));
-            } else {
-              builder = builder.withLoadBalancingPolicy((DCAwareRoundRobinPolicy.builder().build()));
-            }
-          }
-          break;
-        }
-        case "TokenAwareRoundRobinPolicy":
-          builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()));
-          break;
-        case "TokenAwareDCAwareRoundRobinPolicy": {
-          String dataCenter = properties.getProperty(CassandraStoreParameters.DATA_CENTER);
-          boolean allowRemoteDCsForLocalConsistencyLevel = Boolean.parseBoolean(
-                  properties.getProperty(CassandraStoreParameters.ALLOW_REMOTE_DCS_FOR_LOCAL_CONSISTENCY_LEVEL));
-          if (dataCenter != null && !dataCenter.isEmpty()) {
-            if (allowRemoteDCsForLocalConsistencyLevel) {
-              builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(
-                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter)
-                              .allowRemoteDCsForLocalConsistencyLevel().build()));
-            } else {
-              builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(
-                      DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build()));
-            }
-          } else {
-            if (allowRemoteDCsForLocalConsistencyLevel) {
-              builder = builder.withLoadBalancingPolicy(new TokenAwarePolicy(
-                      DCAwareRoundRobinPolicy.builder().allowRemoteDCsForLocalConsistencyLevel().build()));
-            } else {
-              builder = builder.withLoadBalancingPolicy(
-                      new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().build()));
-            }
-          }
-          break;
-        }
-        default:
-          LOG.error("Unsupported Cassandra load balancing  policy: {} ", loadBalancingProp);
-          break;
-      }
-    }
-    return builder;
-  }
-
-  private Cluster.Builder populateCredentials(Properties properties, Cluster.Builder builder)
{
-    String usernameProp = properties.getProperty(CassandraStoreParameters.USERNAME);
-    String passwordProp = properties.getProperty(CassandraStoreParameters.PASSWORD);
-    if (usernameProp != null) {
-      builder = builder.withCredentials(usernameProp, passwordProp);
-    }
-    return builder;
-  }
-
-  private Cluster.Builder populatePoolingSettings(Properties properties, Cluster.Builder
builder) {
-    String localCoreConnectionsPerHost = properties.getProperty(CassandraStoreParameters.LOCAL_CORE_CONNECTIONS_PER_HOST);
-    String remoteCoreConnectionsPerHost = properties.getProperty(CassandraStoreParameters.REMOTE_CORE_CONNECTIONS_PER_HOST);
-    String localMaxConnectionsPerHost = properties.getProperty(CassandraStoreParameters.LOCAL_MAX_CONNECTIONS_PER_HOST);
-    String remoteMaxConnectionsPerHost = properties.getProperty(CassandraStoreParameters.REMOTE_MAX_CONNECTIONS_PER_HOST);
-    String localNewConnectionThreshold = properties.getProperty(CassandraStoreParameters.LOCAL_NEW_CONNECTION_THRESHOLD);
-    String remoteNewConnectionThreshold = properties.getProperty(CassandraStoreParameters.REMOTE_NEW_CONNECTION_THRESHOLD);
-    String localMaxRequestsPerConnection = properties.getProperty(CassandraStoreParameters.LOCAL_MAX_REQUESTS_PER_CONNECTION);
-    String remoteMaxRequestsPerConnection = properties.getProperty(CassandraStoreParameters.REMOTE_MAX_REQUESTS_PER_CONNECTION);
-    PoolingOptions options = new PoolingOptions();
-    if (localCoreConnectionsPerHost != null) {
-      options.setCoreConnectionsPerHost(HostDistance.LOCAL, Integer.parseInt(localCoreConnectionsPerHost));
-    }
-    if (remoteCoreConnectionsPerHost != null) {
-      options.setCoreConnectionsPerHost(HostDistance.REMOTE, Integer.parseInt(remoteCoreConnectionsPerHost));
-    }
-    if (localMaxConnectionsPerHost != null) {
-      options.setMaxConnectionsPerHost(HostDistance.LOCAL, Integer.parseInt(localMaxConnectionsPerHost));
-    }
-    if (remoteMaxConnectionsPerHost != null) {
-      options.setMaxConnectionsPerHost(HostDistance.REMOTE, Integer.parseInt(remoteMaxConnectionsPerHost));
-    }
-    if (localNewConnectionThreshold != null) {
-      options.setNewConnectionThreshold(HostDistance.LOCAL, Integer.parseInt(localNewConnectionThreshold));
-    }
-    if (remoteNewConnectionThreshold != null) {
-      options.setNewConnectionThreshold(HostDistance.REMOTE, Integer.parseInt(remoteNewConnectionThreshold));
-    }
-    if (localMaxRequestsPerConnection != null) {
-      options.setMaxRequestsPerConnection(HostDistance.LOCAL, Integer.parseInt(localMaxRequestsPerConnection));
-    }
-    if (remoteMaxRequestsPerConnection != null) {
-      options.setMaxRequestsPerConnection(HostDistance.REMOTE, Integer.parseInt(remoteMaxRequestsPerConnection));
-    }
-    builder = builder.withPoolingOptions(options);
-    return builder;
-  }
-
-  private Cluster.Builder populateQueryOptions(Properties properties, Cluster.Builder builder)
{
-    String consistencyLevelProp = properties.getProperty(CassandraStoreParameters.CONSISTENCY_LEVEL);
-    String serialConsistencyLevelProp = properties.getProperty(CassandraStoreParameters.SERIAL_CONSISTENCY_LEVEL);
-    String fetchSize = properties.getProperty(CassandraStoreParameters.FETCH_SIZE);
-    QueryOptions options = new QueryOptions();
-    if (consistencyLevelProp != null) {
-      options.setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevelProp));
-    }
-    if (serialConsistencyLevelProp != null) {
-      options.setSerialConsistencyLevel(ConsistencyLevel.valueOf(serialConsistencyLevelProp));
-    }
-    if (fetchSize != null) {
-      options.setFetchSize(Integer.parseInt(fetchSize));
-    }
-    return builder.withQueryOptions(options);
-  }
-
-  private Cluster.Builder populateReconnectPolicy(Properties properties, Cluster.Builder
builder) {
-    String reconnectionPolicy = properties.getProperty(CassandraStoreParameters.RECONNECTION_POLICY);
-    if (reconnectionPolicy != null) {
-      switch (reconnectionPolicy) {
-        case "ConstantReconnectionPolicy": {
-          String constantReconnectionPolicyDelay = properties.getProperty(CassandraStoreParameters.CONSTANT_RECONNECTION_POLICY_DELAY);
-          ConstantReconnectionPolicy policy = new ConstantReconnectionPolicy(Long.parseLong(constantReconnectionPolicyDelay));
-          builder = builder.withReconnectionPolicy(policy);
-          break;
-        }
-        case "ExponentialReconnectionPolicy": {
-          String exponentialReconnectionPolicyBaseDelay = properties.getProperty(CassandraStoreParameters.EXPONENTIAL_RECONNECTION_POLICY_BASE_DELAY);
-          String exponentialReconnectionPolicyMaxDelay = properties.getProperty(CassandraStoreParameters.EXPONENTIAL_RECONNECTION_POLICY_MAX_DELAY);
-
-          ExponentialReconnectionPolicy policy = new ExponentialReconnectionPolicy(Long.parseLong(exponentialReconnectionPolicyBaseDelay),
-                  Long.parseLong(exponentialReconnectionPolicyMaxDelay));
-          builder = builder.withReconnectionPolicy(policy);
-          break;
-        }
-        default:
-          LOG.error("Unsupported reconnection policy : {} ", reconnectionPolicy);
-      }
-    }
-    return builder;
-  }
-
-  private Cluster.Builder populateRetrytPolicy(Properties properties, Cluster.Builder builder)
{
-    String retryPolicy = properties.getProperty(CassandraStoreParameters.RETRY_POLICY);
-    if (retryPolicy != null) {
-      switch (retryPolicy) {
-        case "DefaultRetryPolicy":
-          builder = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE);
-          break;
-        case "DowngradingConsistencyRetryPolicy":
-          builder = builder.withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE);
-          break;
-        case "FallthroughRetryPolicy":
-          builder = builder.withRetryPolicy(FallthroughRetryPolicy.INSTANCE);
-          break;
-        case "LoggingDefaultRetryPolicy":
-          builder = builder.withRetryPolicy(new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE));
-          break;
-        case "LoggingDowngradingConsistencyRetryPolicy":
-          builder = builder.withRetryPolicy(new LoggingRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE));
-          break;
-        case "LoggingFallthroughRetryPolicy":
-          builder = builder.withRetryPolicy(new LoggingRetryPolicy(FallthroughRetryPolicy.INSTANCE));
-          break;
-        default:
-          LOG.error("Unsupported retry policy : {} ", retryPolicy);
-          break;
-      }
-    }
-    return builder;
-  }
-
-  private Cluster.Builder populateSocketOptions(Properties properties, Cluster.Builder builder)
{
-    String connectionTimeoutMillisProp = properties.getProperty(CassandraStoreParameters.CONNECTION_TIMEOUT_MILLIS);
-    String keepAliveProp = properties.getProperty(CassandraStoreParameters.KEEP_ALIVE);
-    String readTimeoutMillisProp = properties.getProperty(CassandraStoreParameters.READ_TIMEOUT_MILLIS);
-    String receiveBufferSizeProp = properties.getProperty(CassandraStoreParameters.RECEIVER_BUFFER_SIZE);
-    String reuseAddress = properties.getProperty(CassandraStoreParameters.REUSE_ADDRESS);
-    String sendBufferSize = properties.getProperty(CassandraStoreParameters.SEND_BUFFER_SIZE);
-    String soLinger = properties.getProperty(CassandraStoreParameters.SO_LINGER);
-    String tcpNoDelay = properties.getProperty(CassandraStoreParameters.TCP_NODELAY);
-    SocketOptions options = new SocketOptions();
-    if (connectionTimeoutMillisProp != null) {
-      options.setConnectTimeoutMillis(Integer.parseInt(connectionTimeoutMillisProp));
-    }
-    if (keepAliveProp != null) {
-      options.setKeepAlive(Boolean.parseBoolean(keepAliveProp));
-    }
-    if (readTimeoutMillisProp != null) {
-      options.setReadTimeoutMillis(Integer.parseInt(readTimeoutMillisProp));
-    }
-    if (receiveBufferSizeProp != null) {
-      options.setReceiveBufferSize(Integer.parseInt(receiveBufferSizeProp));
-    }
-    if (reuseAddress != null) {
-      options.setReuseAddress(Boolean.parseBoolean(reuseAddress));
-    }
-    if (sendBufferSize != null) {
-      options.setSendBufferSize(Integer.parseInt(sendBufferSize));
-    }
-    if (soLinger != null) {
-      options.setSoLinger(Integer.parseInt(soLinger));
-    }
-    if (tcpNoDelay != null) {
-      options.setTcpNoDelay(Boolean.parseBoolean(tcpNoDelay));
-    }
-    return builder.withSocketOptions(options);
-  }
-
   /**
    * {@inheritDoc}
    * <p>
    * This is a setter method to set the class of persistent objects.
    *
    * @param persistentClass class of persistent objects
-   *                        {@link org.apache.gora.cassandra.serializers.CassandraNativePersistent}
+   *                        {@link CassandraNativePersistent}
    *                        {@link  org.apache.gora.persistency.Persistent}
    */
   @Override
@@ -638,9 +109,10 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
     this.persistentClass = persistentClass;
   }
 
+  @SuppressWarnings("all")
   @Override
   public Class<T> getPersistentClass() {
-    return this.persistentClass;
+    return (Class<T>) this.persistentClass;
   }
 
   @Override
@@ -650,20 +122,15 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
 
   @Override
   public void createSchema() {
-    LOG.debug("creating Cassandra keyspace {}", mapping.getKeySpace().getName());
-    this.session.execute(CassandraQueryFactory.getCreateKeySpaceQuery(mapping));
-    LOG.debug("creating Cassandra column family / table {}", mapping.getCoreName());
-    this.session.execute(CassandraQueryFactory.getCreateTableQuery(mapping));
+    cassandraSerializer.createSchema();
   }
 
   @Override
   public void deleteSchema() {
-    LOG.debug("dropping Cassandra table {}", mapping.getCoreName());
-    this.session.execute(CassandraQueryFactory.getDropTableQuery(mapping));
-    LOG.debug("dropping Cassandra keyspace {}", mapping.getKeySpace().getName());
-    this.session.execute(CassandraQueryFactory.getDropKeySpaceQuery(mapping));
+    cassandraSerializer.deleteSchema();
   }
 
+  @SuppressWarnings("all")
   @Override
   public Class<K> getKeyClass() {
     return this.keyClass;
@@ -677,16 +144,32 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
   @Override
   public K newKey() {
     try {
-      return beanFactory.newKey();
+      if (beanFactory != null) {
+        return beanFactory.newKey();
+      } else {
+        LOG.warn("beanFactory is hasn't been initialized. It's recommended to initialize
beanFactory.");
+        return keyClass.newInstance();
+      }
     } catch (Exception ex) {
       LOG.error(ex.getMessage(), ex);
       return null;
     }
   }
 
+  @SuppressWarnings("all")
   @Override
   public T newPersistent() {
-    return this.beanFactory.newPersistent();
+    try {
+      if (beanFactory != null) {
+        return this.beanFactory.newPersistent();
+      } else {
+        LOG.warn("beanFactory is hasn't been initialized. It's recommended to initialize
beanFactory.");
+        return persistentClass.newInstance();
+      }
+    } catch (Exception ex) {
+      LOG.error(ex.getMessage(), ex);
+      return null;
+    }
   }
 
   @Override
@@ -701,17 +184,12 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
 
   @Override
   public void close() {
-    this.session.close();
-    this.cluster.close();
+    this.cassandraSerializer.close();
   }
 
   @Override
   public T get(K key) {
-    if (isUseNativeSerialization) {
-      return mapper.get(key);
-    } else {
-      return null;
-    }
+    return (T) cassandraSerializer.get(key);
   }
 
   @Override
@@ -721,21 +199,12 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
 
   @Override
   public void put(K key, T obj) {
-    if (isUseNativeSerialization) {
-      mapper.save(obj);
-    } else {
-
-    }
+    cassandraSerializer.put(key, obj);
   }
 
   @Override
   public boolean delete(K key) {
-    if (isUseNativeSerialization) {
-      mapper.delete(key);
-      return true;
-    } else {
-      return false;
-    }
+    return cassandraSerializer.delete(key);
   }
 
   @Override
@@ -760,7 +229,7 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
 
   @Override
   public void flush() {
-
+ // ignore since caching has been disabled
   }
 
   @Override
@@ -770,19 +239,12 @@ public class CassandraStore<K, T extends Persistent> implements
DataStore<K, T>
 
   @Override
   public void truncateSchema() {
-    LOG.debug("truncating Cassandra table {}", mapping.getCoreName());
-    this.session.execute(CassandraQueryFactory.getTruncateTableQuery(mapping));
+    cassandraSerializer.truncateSchema();
   }
 
   @Override
   public boolean schemaExists() {
-    KeyspaceMetadata keyspace = cluster.getMetadata().getKeyspace(mapping.getKeySpace().getName());
-    if (keyspace != null) {
-      TableMetadata table = keyspace.getTable(mapping.getCoreName());
-      return table != null;
-    } else {
-      return false;
-    }
+    return cassandraSerializer.schemaExists();
   }
 
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStoreParameters.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStoreParameters.java
b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStoreParameters.java
index c48b1c0..95e1c0f 100644
--- a/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStoreParameters.java
+++ b/gora-cassandra-cql/src/main/java/org/apache/gora/cassandra/store/CassandraStoreParameters.java
@@ -218,7 +218,7 @@ public class CassandraStoreParameters {
    * Property pointing to use Native Cassandra Native Serialization.
    * boolean
    */
-  public static final String USE_CASSANDRA_NATIVE_SERIALIZATION = "gora.cassandrastore.useCassandraNativeSerialization";
+  public static final String CASSANDRA_SERIALIZATION_TYPE = "gora.cassandrastore.cassandraSerializationType";
   /**
    * Property pointing to the custom codec file.
    * string

http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/test/conf/avro/gora-cassandra-mapping.xml
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/test/conf/avro/gora-cassandra-mapping.xml b/gora-cassandra-cql/src/test/conf/avro/gora-cassandra-mapping.xml
new file mode 100644
index 0000000..e99cf92
--- /dev/null
+++ b/gora-cassandra-cql/src/test/conf/avro/gora-cassandra-mapping.xml
@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+   Licensed to the Apache Software Foundation (ASF) under one or more
+   contributor license agreements.  See the NOTICE file distributed with
+   this work for additional information regarding copyright ownership.
+   The ASF licenses this file to You 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.
+-->
+
+<!--
+   The value of 'host' attribute of keyspace tag should match exactly what is in
+   gora.properties file. Essentially this means that if you are using port number, you should
+   use it every where regardless of whether it is the default port or not.
+   At runtime Gora will otherwise try to connect to localhost
+   https://issues.apache.org/jira/browse/GORA-269
+
+   The values of 'replication_factor' and 'placement_strategy' attribute of keyspace tag
+   only apply if gora create the kyespace. they have no effect if this is being used against

+   an existing keyspace. the default value for 'replication_factor' is '1'
+   
+   The value of 'placement_strategy' should be a fully qualifed class name that is known
to
+   the cassansra cluster, not the application or gora. As of this writing, the classes that
ship
+   with cassandra are:
+   'org.apache.cassandra.locator.SimpleStrategy'
+   'org.apache.cassandra.locator.NetworkTopologyStrategy'
+   gora cassandra would use SimpleStrategy by default if no value for this attribute is specified
+   
+   The default value of 'gc_grace_seconds' is '0' which is ONLY VIABLE FOR SINGLE NODE
+   CLUSTER. you should update this value according to your cluster configuration. 
+   https://wiki.apache.org/cassandra/StorageConfiguration
+
+   The value of 'ttl' (time to live) attribute of field tag should most likely always
+   be zero unless you want Cassandra to create Tombstones and delete portions of your
+   data once this period expires. Any positive value is read and bound to the number
+   of seconds after which the value for that field will disappear. The default value of ttl
+   is '0'
+
+   More information on gora-cassandra configuration and mapping's can be found
+   at http://gora.apache.org/current/gora-cassandra.html
+-->
+
+<gora-otd>
+
+  <keyspace name="avroKeySpace" durableWrite="false">
+    <placementStrategy name="SimpleStrategy" replication_factor="1"/>
+  </keyspace>
+  
+  <class name="org.apache.gora.examples.generated.WebPage" keyClass="java.lang.String"
table="WebPage" keyspace="avroKeySpace">
+    <field name="url" column="url" length="128" primarykey="true"/>
+    <field name="content" column="content" type="blob"/>
+    <field name="parsedContent" column="parsedContent" type="list"/>
+    <field name="outlinks" column="outlinks" type="map"/>
+  </class>
+
+  <class name="org.apache.gora.examples.generated.Employee" keyClass="java.lang.String"
keyspace="avroKeySpace"
+         table="Employee" compactStorage="true" >
+    <field name="name" column="name" type="text" ttl="10"/>
+    <field name="dateOfBirth" column="dob" type="bigint" ttl="10"/>
+    <field name="ssn" column="ssn" type="int" ttl="10" primarykey="true"/>
+    <field name="salary" column="salary" type="int" ttl="10" />
+   </class>
+
+</gora-otd>

http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/test/conf/nativeSerialization/gora-cassandra-mapping.xml
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/test/conf/nativeSerialization/gora-cassandra-mapping.xml
b/gora-cassandra-cql/src/test/conf/nativeSerialization/gora-cassandra-mapping.xml
index 07d6590..d3dde32 100644
--- a/gora-cassandra-cql/src/test/conf/nativeSerialization/gora-cassandra-mapping.xml
+++ b/gora-cassandra-cql/src/test/conf/nativeSerialization/gora-cassandra-mapping.xml
@@ -55,7 +55,7 @@
         <placementStrategy name="SimpleStrategy" replication_factor="1"/>
     </keyspace>
 
-    <class name="org.apache.gora.cassandra.test.nativeSerialization.User" keyClass="java.util.UUID"
keyspace="nativeTestKeySpace"
+    <class name="org.apache.gora.cassandra.example.generated.nativeSerialization.User"
keyClass="java.util.UUID" keyspace="nativeTestKeySpace"
            table="Users" compactStorage="true" >
         <field name="userId" column="user_id" type="uuid" ttl="10" primarykey="true"/>
         <field name="name" column="name" type="text" ttl="10"/>

http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/test/java/.gitignore
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/test/java/.gitignore b/gora-cassandra-cql/src/test/java/.gitignore
deleted file mode 100644
index 09697dc..0000000
--- a/gora-cassandra-cql/src/test/java/.gitignore
+++ /dev/null
@@ -1,15 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You 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.
-

http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/store/TestCassandraStoreWithNativeSerialization.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/store/TestCassandraStoreWithNativeSerialization.java
b/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/store/TestCassandraStoreWithNativeSerialization.java
index 5bb1114..9ebcae5 100644
--- a/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/store/TestCassandraStoreWithNativeSerialization.java
+++ b/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/store/TestCassandraStoreWithNativeSerialization.java
@@ -1,7 +1,24 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You 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 org.apache.gora.cassandra.store;
 
 import org.apache.gora.cassandra.GoraCassandraTestDriver;
-import org.apache.gora.cassandra.test.nativeSerialization.User;
+import org.apache.gora.cassandra.example.generated.nativeSerialization.User;
 import org.apache.gora.store.DataStore;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -34,7 +51,7 @@ public class TestCassandraStoreWithNativeSerialization {
     parameter = new Properties();
     parameter.setProperty(CassandraStoreParameters.CASSANDRA_SERVERS, "localhost");
     parameter.setProperty(CassandraStoreParameters.PORT, "9042");
-    parameter.setProperty(CassandraStoreParameters.USE_CASSANDRA_NATIVE_SERIALIZATION, "true");
+    parameter.setProperty(CassandraStoreParameters.CASSANDRA_SERIALIZATION_TYPE, "native");
     parameter.setProperty(CassandraStoreParameters.PROTOCOL_VERSION, "3");
     parameter.setProperty(CassandraStoreParameters.CLUSTER_NAME,"Test Cluster");
     parameter.setProperty("gora.cassandrastore.mapping.file", "nativeSerialization/gora-cassandra-mapping.xml");
@@ -75,11 +92,11 @@ public class TestCassandraStoreWithNativeSerialization {
     // storing data;
     userDataStore.put(id, user1);
     // get data;
-    User olduser =  userDataStore.get(id);
+    User olduser =  userDataStore.get(user1.getUserId());
     Assert.assertEquals(olduser.getName(), user1.getName());
     Assert.assertEquals(olduser.getDateOfBirth(), user1.getDateOfBirth());
     // delete data;
-    userDataStore.delete(id);
+    userDataStore.delete(user1.getUserId());
     // get data
     User deletedUser = userDataStore.get(id);
     Assert.assertNull(deletedUser);

http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/DateAsStringCodec.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/DateAsStringCodec.java
b/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/DateAsStringCodec.java
index f6061cf..b5ffedb 100644
--- a/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/DateAsStringCodec.java
+++ b/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/DateAsStringCodec.java
@@ -1,3 +1,20 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You 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 org.apache.gora.cassandra.test.nativeSerialization;
 
 import com.datastax.driver.core.TypeCodec;

http://git-wip-us.apache.org/repos/asf/gora/blob/5e383ef9/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/User.java
----------------------------------------------------------------------
diff --git a/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/User.java
b/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/User.java
deleted file mode 100644
index 0f4c7ee..0000000
--- a/gora-cassandra-cql/src/test/java/org/apache/gora/cassandra/test/nativeSerialization/User.java
+++ /dev/null
@@ -1,66 +0,0 @@
-package org.apache.gora.cassandra.test.nativeSerialization;
-
-import com.datastax.driver.mapping.annotations.Column;
-import com.datastax.driver.mapping.annotations.PartitionKey;
-import com.datastax.driver.mapping.annotations.Table;
-import com.datastax.driver.mapping.annotations.Transient;
-import org.apache.gora.cassandra.serializers.CassandraNativePersistent;
-
-import java.util.Date;
-import java.util.UUID;
-
-/**
- * Created by madhawa on 6/23/17.
- */
-
-@Table(keyspace = "nativeTestKeySpace", name = "users",
-        readConsistency = "QUORUM",
-        writeConsistency = "QUORUM",
-        caseSensitiveKeyspace = false,
-        caseSensitiveTable = false)
-public class User extends CassandraNativePersistent {
-  @PartitionKey
-  @Column(name = "user_id")
-  private UUID userId;
-  @Column(name = "name")
-  private String name;
-  @Column(name = "dob")
-  private Date dateOfBirth;
-
-  @Transient
-  private boolean dirty;
-
-  public User() {
-
-  }
-
-  public User(UUID userId, String name, Date dateOfBirth) {
-    this.userId = userId;
-    this.name = name;
-    this.dateOfBirth = dateOfBirth;
-  }
-
-  public void setUserId(UUID userId) {
-    this.userId = userId;
-  }
-
-  public void setName(String name) {
-    this.name = name;
-  }
-
-  public void setDateOfBirth(Date dateOfBirth) {
-    this.dateOfBirth = dateOfBirth;
-  }
-
-  public UUID getUserId() {
-    return userId;
-  }
-
-  public String getName() {
-    return name;
-  }
-
-  public Date getDateOfBirth() {
-    return dateOfBirth;
-  }
-}


Mime
View raw message