Merge pull request #151 from Stratio/fix/DECISION-271_messages_with_n…

…ull_values [DECISION-271] Deal with null fields in Avro serialization
Stratio · Mar 31, 2016 · e477d92 · e477d92
2 parents eb384e5 + ae9ff6a
commit e477d92
Show file tree

Hide file tree

Showing 6 changed files with 120 additions and 107 deletions.
diff --git a/api/README.md b/api/README.md
@@ -23,7 +23,7 @@ Requirements
   * Scala 2.10.3
   * sbt 0.13
   * zookeeper
-  * kafka 0.8.1
+  * kafka 0.8.2.1
 
 How to start
 ============

diff --git a/api/src/main/scala/com/stratio/decision/api/StreamingAPIAsyncOperation.scala b/api/src/main/scala/com/stratio/decision/api/StreamingAPIAsyncOperation.scala
@@ -41,8 +41,14 @@ class StreamingAPIAsyncOperation(tableProducer: KafkaProducer) {
     var c : ColumnType = null
 
     for (messageColumn : ColumnNameTypeValue <- stratioStreamingMessage.getColumns){
-      c = new ColumnType(messageColumn.getColumn, messageColumn.getValue.toString, messageColumn.getValue.getClass
-        .getName)
+
+      if (messageColumn.getValue != null) {
+        c = new ColumnType(messageColumn.getColumn, messageColumn.getValue.toString, messageColumn.getValue.getClass
+          .getName)
+      } else {
+
+        c = new ColumnType(messageColumn.getColumn, null, null);
+      }
       columns.add(c)
     }
 

diff --git a/commons/src/main/java/com/stratio/decision/commons/avro/Action.java b/commons/src/main/java/com/stratio/decision/commons/avro/Action.java
@@ -13,10 +13,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.stratio.decision.commons.avro;  
+package com.stratio.decision.commons.avro;
 @SuppressWarnings("all")
 @org.apache.avro.specific.AvroGenerated
-public enum Action { 
+public enum Action {
   LISTEN, SAVE_TO_CASSANDRA, SAVE_TO_MONGO, SAVE_TO_SOLR, SAVE_TO_ELASTICSEARCH  ;
   public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"enum\",\"name\":\"Action\",\"namespace\":\"com.stratio.decision.commons.avro\",\"symbols\":[\"LISTEN\",\"SAVE_TO_CASSANDRA\",\"SAVE_TO_MONGO\",\"SAVE_TO_SOLR\",\"SAVE_TO_ELASTICSEARCH\"]}");
   public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }

diff --git a/commons/src/main/java/com/stratio/decision/commons/avro/ColumnType.java b/commons/src/main/java/com/stratio/decision/commons/avro/ColumnType.java
@@ -17,8 +17,8 @@
 @SuppressWarnings("all")
 @org.apache.avro.specific.AvroGenerated
 public class ColumnType extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
-  private static final long serialVersionUID = 6385653382887656308L;
-  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"ColumnType\",\"namespace\":\"com.stratio.decision.commons.avro\",\"fields\":[{\"name\":\"column\",\"type\":\"string\"},{\"name\":\"value\",\"type\":\"string\"},{\"name\":\"type\",\"type\":\"string\"}]}");
+  private static final long serialVersionUID = 5874033527197767193L;
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"ColumnType\",\"namespace\":\"com.stratio.decision.commons.avro\",\"fields\":[{\"name\":\"column\",\"type\":\"string\"},{\"name\":\"value\",\"type\":[\"null\",\"string\"],\"default\":\"null\"},{\"name\":\"type\",\"type\":[\"null\",\"string\"],\"default\":\"null\"}]}");
   public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
   @Deprecated public java.lang.CharSequence column;
   @Deprecated public java.lang.CharSequence value;
@@ -334,4 +334,4 @@ public ColumnType build() {
     READER$.read(this, org.apache.avro.specific.SpecificData.getDecoder(in));
   }
 
-}
+}