Merge branch 'server-transforms' into server-transforms-network

src/java/voldemort/store/StoreDefinition.java

@@ -20,7 +20,6 @@
 
 import voldemort.client.RoutingTier;
 import voldemort.serialization.SerializerDefinition;
-import voldemort.serialization.SerializerFactory;
 import voldemort.store.views.View;
 import voldemort.utils.Utils;
 
@@ -51,7 +50,7 @@ public class StoreDefinition implements Serializable {
     private final String routingStrategyType;
     private final String viewOf;
     private final View<?, ?, ?, ?> valueTransformation;
-    private final SerializerFactory serializerFactory;
+    private final String serializerFactory;
 
     public StoreDefinition(String name,
                            String type,
@@ -69,7 +68,7 @@ public StoreDefinition(String name,
                            View<?, ?, ?, ?> valTrans,
                            Integer retentionDays,
                            Integer retentionThrottleRate,
-                           SerializerFactory factory) {
+                           String factory) {
         this.name = Utils.notNull(name);
         this.type = Utils.notNull(type);
         this.replicationFactor = replicationFactor;
@@ -210,7 +209,7 @@ public boolean hasValueTransformation() {
         return valueTransformation;
     }
 
-    public SerializerFactory getSerializerFactory() {
+    public String getSerializerFactory() {
         return this.serializerFactory;
     }
 
@@ -245,10 +244,9 @@ && getRoutingPolicy() == def.getRoutingPolicy()
                                                                           .getClass() : null)
                && Objects.equal(getRetentionDays(), def.getRetentionDays())
                && Objects.equal(getRetentionScanThrottleRate(), def.getRetentionScanThrottleRate())
-               && Objects.equal(getSerializerFactory() != null ? getSerializerFactory().getClass()
-                                                              : null,
+               && Objects.equal(getSerializerFactory() != null ? getSerializerFactory() : null,
                                 def.getSerializerFactory() != null ? def.getSerializerFactory()
-                                                                        .getClass() : null);
+                                                                  : null);
     }
 
     @Override

src/java/voldemort/store/StoreDefinitionBuilder.java

@@ -2,7 +2,6 @@
 
 import voldemort.client.RoutingTier;
 import voldemort.serialization.SerializerDefinition;
-import voldemort.serialization.SerializerFactory;
 import voldemort.store.views.View;
 import voldemort.utils.Utils;
 
@@ -30,7 +29,7 @@ public class StoreDefinitionBuilder {
     private String routingStrategyType = null;
     private String viewOf = null;
     private View<?, ?, ?, ?> view = null;
-    private SerializerFactory serializerFactory = null;
+    private String serializerFactory = null;
 
     public String getName() {
         return Utils.notNull(name);
@@ -192,11 +191,11 @@ public StoreDefinitionBuilder setView(View<?, ?, ?, ?> valueTransformation) {
         return this;
     }
 
-    public SerializerFactory getSerializerFactory() {
+    public String getSerializerFactory() {
         return this.serializerFactory;
     }
 
-    public StoreDefinitionBuilder setSerializerFactory(SerializerFactory factory) {
+    public StoreDefinitionBuilder setSerializerFactory(String factory) {
         this.serializerFactory = factory;
         return this;
     }

src/java/voldemort/store/views/ViewStorageConfiguration.java

@@ -14,6 +14,7 @@
 import voldemort.store.compress.CompressionStrategy;
 import voldemort.store.compress.CompressionStrategyFactory;
 import voldemort.utils.ByteArray;
+import voldemort.utils.ReflectUtils;
 import voldemort.utils.Utils;
 
 public class ViewStorageConfiguration implements StorageConfiguration {
@@ -41,9 +42,13 @@ public StorageEngine<ByteArray, byte[], byte[]> getStore(String name) {
         if(target == null)
             throw new VoldemortException("View \"" + name + "\" has a target store \"" + targetName
                                          + "\" which does not exist.");
-        SerializerFactory factory = def.getSerializerFactory();
-        if(factory == null)
+        String factoryName = def.getSerializerFactory();
+        SerializerFactory factory;
+        if(factoryName == null)
             factory = new DefaultSerializerFactory();
+        else
+            factory = loadSerializerFactory(factoryName);
+
         /* Check if the values in the target store are compressed */
         CompressionStrategy valueCompressionStrategy = null;
         if(targetDef.getValueSerializer().hasCompression()) {
@@ -65,4 +70,11 @@ public String getType() {
         return TYPE_NAME;
     }
 
+    private SerializerFactory loadSerializerFactory(String className) {
+        if(className == null)
+            return null;
+        Class<?> transClass = ReflectUtils.loadClass(className.trim());
+        return (SerializerFactory) ReflectUtils.callConstructor(transClass, new Object[] {});
+    }
+
 }

src/java/voldemort/xml/StoreDefinitionsMapper.java

@@ -45,7 +45,6 @@
 import voldemort.routing.RoutingStrategyType;
 import voldemort.serialization.Compression;
 import voldemort.serialization.SerializerDefinition;
-import voldemort.serialization.SerializerFactory;
 import voldemort.store.StoreDefinition;
 import voldemort.store.StoreDefinitionBuilder;
 import voldemort.store.StoreUtils;
@@ -219,10 +218,9 @@ private StoreDefinition readView(Element store, List<StoreDefinition> stores) {
                                                   STORE_PREFERRED_WRITES_ELMT,
                                                   target.getRequiredReads());
 
-        SerializerFactory viewSerializerFactory = null;
+        String viewSerializerFactoryName = null;
         if(store.getChildText(VIEW_SERIALIZER_FACTORY_ELMT) != null) {
-            String serializerFactoryName = store.getChild(VIEW_SERIALIZER_FACTORY_ELMT).getText();
-            viewSerializerFactory = loadSerializerFactory(serializerFactoryName);
+            viewSerializerFactoryName = store.getChild(VIEW_SERIALIZER_FACTORY_ELMT).getText();
         }
 
         SerializerDefinition keySerializer = target.getKeySerializer();
@@ -255,7 +253,7 @@ private StoreDefinition readView(Element store, List<StoreDefinition> stores) {
                                            .setPreferredWrites(preferredWrites)
                                            .setRequiredWrites(requiredWrites)
                                            .setView(valTrans)
-                                           .setSerializerFactory(viewSerializerFactory)
+                                           .setSerializerFactory(viewSerializerFactoryName)
                                            .build();
     }
 
@@ -266,13 +264,6 @@ private StoreDefinition readView(Element store, List<StoreDefinition> stores) {
         return (View<?, ?, ?, ?>) ReflectUtils.callConstructor(transClass, new Object[] {});
     }
 
-    private SerializerFactory loadSerializerFactory(String className) {
-        if(className == null)
-            return null;
-        Class<?> transClass = ReflectUtils.loadClass(className.trim());
-        return (SerializerFactory) ReflectUtils.callConstructor(transClass, new Object[] {});
-    }
-
     private SerializerDefinition readSerializer(Element elmt) {
         String name = elmt.getChild(STORE_SERIALIZATION_TYPE_ELMT).getText();
         boolean hasVersion = true;
@@ -388,7 +379,7 @@ private Element viewToElement(StoreDefinition storeDefinition) {
 
         Element serializerFactory = new Element(VIEW_SERIALIZER_FACTORY_ELMT);
         if(storeDefinition.getSerializerFactory() != null) {
-            serializerFactory.setText(storeDefinition.getSerializerFactory().getClass().getName());
+            serializerFactory.setText(storeDefinition.getSerializerFactory());
             store.addContent(serializerFactory);
         }
         return store;