Google Git
Sign in
apache/samza/samza-sql^!/.
commit
a7de73594b3b85b3ad4132cc33dff1f5efcd95bf
[log] [tgz]
author
Yi Pan (Data Infrastructure) <nickpan47@gmail.com>
Tue Nov 22 15:39:27 2016 -0800
committer
Yi Pan (Data Infrastructure) <nickpan47@gmail.com>
Tue Nov 22 15:39:27 2016 -0800
tree
c5a711e1afe80f289f11559ac70ad9271195f55e
parent
1dac25e1750b7f3fefa72def855136801462494d [diff]
SAMZA-1050: Make samza-operator independent of avro version

Prep for merging the samza-operator APIs to master: removing the direct dependency on avro.

Author: Yi Pan (Data Infrastructure) <nickpan47@gmail.com>

Reviewers: Jagadish <jagadish1989@gmail.com>, Prateek Maheshiwari <pmaheshiwari@linkedin.com>

Closes #22 from nickpan47/SAMZA-1050

diff --git a/build.gradle b/build.gradle
index 28c2dcf..1b3c278 100644
--- a/build.gradle
+++ b/build.gradle

@@ -278,7 +278,12 @@
 
   // Force scala joint compilation
   sourceSets.main.scala.srcDir "src/main/java"
+  sourceSets.test.scala.srcDir "src/test/java"
+
+  // Disable the Javac compiler by forcing joint compilation by scalac. This is equivalent to setting
+  // tasks.compileTestJava.enabled = false
   sourceSets.main.java.srcDirs = []
+  sourceSets.test.java.srcDirs = []
 
   dependencies {
     compile project(':samza-api')
@@ -304,6 +309,10 @@
       // Exclude because YARN's 3.4.5 ZK version is incompatbile with Kafka's 3.3.4.
       exclude module: 'zookeeper'
     }
+    compile("org.apache.hadoop:hadoop-hdfs:$yarnVersion") {
+      exclude module: 'slf4j-log4j12'
+      exclude module: 'servlet-api'
+    }
     compile("org.scalatra:scalatra_$scalaVersion:$scalatraVersion") {
       exclude module: 'scala-compiler'
       exclude module: 'slf4j-api'
@@ -317,6 +326,7 @@
     testCompile "junit:junit:$junitVersion"
     testCompile "org.mockito:mockito-all:$mockitoVersion"
     testCompile project(":samza-core_$scalaVersion").sourceSets.test.output
+    testCompile "org.scalatest:scalatest_$scalaVersion:$scalaTestVersion"
   }
 
   repositories {
@@ -346,7 +356,6 @@
       compile project(":samza-core_$scalaVersion")
       compile "commons-collections:commons-collections:$commonsCollectionVersion"
       compile "org.apache.commons:commons-lang3:$commonsLang3Version"
-      compile "org.apache.avro:avro:$avroVersion"
       compile "org.reactivestreams:reactive-streams:$reactiveStreamVersion"
 
       testCompile project(":samza-api").sourceSets.test.output
@@ -368,6 +377,7 @@
 
     dependencies {
       compile project(":samza-operator")
+      compile "org.apache.avro:avro:$avroVersion"
       compile "org.apache.calcite:calcite-core:$calciteVersion"
       testCompile "junit:junit:$junitVersion"
       testCompile "org.mockito:mockito-all:$mockitoVersion"
@@ -470,7 +480,13 @@
 project(":samza-kv-rocksdb_$scalaVersion") {
   apply plugin: 'scala'
 
+  // Force scala joint compilation
+  sourceSets.main.scala.srcDir "src/main/java"
   sourceSets.test.scala.srcDir "src/test/java"
+
+  // Disable the Javac compiler by forcing joint compilation by scalac. This is equivalent to setting
+  // tasks.compileTestJava.enabled = false
+  sourceSets.main.java.srcDirs = []
   sourceSets.test.java.srcDirs = []
 
   dependencies {

diff --git a/samza-operator/src/test/java/org/apache/samza/task/BroadcastOperatorTask.java b/samza-operator/src/test/java/org/apache/samza/task/BroadcastOperatorTask.java
index 493a688..a6d57da 100644
--- a/samza-operator/src/test/java/org/apache/samza/task/BroadcastOperatorTask.java
+++ b/samza-operator/src/test/java/org/apache/samza/task/BroadcastOperatorTask.java

@@ -19,11 +19,10 @@
 
 package org.apache.samza.task;
 
-import org.apache.avro.generic.GenericRecord;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.MessageStreams.SystemMessageStream;
-import org.apache.samza.operators.Windows;
 import org.apache.samza.operators.TriggerBuilder;
+import org.apache.samza.operators.Windows;
 import org.apache.samza.operators.data.IncomingSystemMessage;
 import org.apache.samza.operators.data.Offset;
 import org.apache.samza.operators.task.StreamOperatorTask;
@@ -82,16 +81,7 @@
   }
 
   JsonMessage getInputMessage(IncomingSystemMessage m1) {
-    return new JsonMessage(
-        m1.getKey().toString(),
-        (MessageType) m1.getMessage(),
-        m1.getOffset(),
-        this.getEventTime((GenericRecord)m1.getMessage()),
-        m1.getSystemStreamPartition());
-  }
-
-  long getEventTime(GenericRecord msg) {
-    return (Long) msg.get("event_time");
+    return (JsonMessage) m1.getMessage();
   }
 
   boolean myFilter1(JsonMessage m1) {

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/api/data/Data.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/Data.java
similarity index 96%
rename from samza-operator/src/main/java/org/apache/samza/operators/api/data/Data.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/Data.java
index 69a3bee..7d6ee79 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/api/data/Data.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/Data.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.operators.api.data;
+package org.apache.samza.sql.calcite.data;
 
 import java.util.List;
 import java.util.Map;

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/api/data/Schema.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/Schema.java
similarity index 96%
rename from samza-operator/src/main/java/org/apache/samza/operators/api/data/Schema.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/Schema.java
index dc3f8f4..e2a79cf 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/api/data/Schema.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/Schema.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.operators.api.data;
+package org.apache.samza.sql.calcite.data;
 
 import java.util.Map;
 

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/avro/AvroData.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/avro/AvroData.java
similarity index 97%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/avro/AvroData.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/avro/AvroData.java
index e4f5d79..91d26a2 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/avro/AvroData.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/avro/AvroData.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.operators.impl.data.avro;
+package org.apache.samza.sql.calcite.data.avro;
 
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -25,8 +25,8 @@
 
 import org.apache.avro.generic.GenericArray;
 import org.apache.avro.generic.GenericRecord;
-import org.apache.samza.operators.api.data.Data;
-import org.apache.samza.operators.api.data.Schema;
+import org.apache.samza.sql.calcite.data.Data;
+import org.apache.samza.sql.calcite.data.Schema;
 
 
 public class AvroData implements Data {

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/avro/AvroSchema.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/avro/AvroSchema.java
similarity index 98%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/avro/AvroSchema.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/avro/AvroSchema.java
index c04e4f6..c3bb150 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/avro/AvroSchema.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/avro/AvroSchema.java

@@ -17,14 +17,14 @@
  * under the License.
  */
 
-package org.apache.samza.operators.impl.data.avro;
+package org.apache.samza.sql.calcite.data.avro;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.avro.Schema.Field;
-import org.apache.samza.operators.api.data.Data;
-import org.apache.samza.operators.api.data.Schema;
+import org.apache.samza.sql.calcite.data.Data;
+import org.apache.samza.sql.calcite.data.Schema;
 
 
 public class AvroSchema implements Schema {

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerde.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerde.java
similarity index 94%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerde.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerde.java
index 2432aca..97a3b6c 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerde.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerde.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.impl.data.serializers;
+package org.apache.samza.sql.calcite.data.serializers;
 
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericDatumReader;
@@ -27,8 +27,8 @@
 import org.apache.avro.io.BinaryEncoder;
 import org.apache.samza.SamzaException;
 import org.apache.samza.serializers.Serde;
-import org.apache.samza.operators.impl.data.avro.AvroData;
-import org.apache.samza.operators.impl.data.avro.AvroSchema;
+import org.apache.samza.sql.calcite.data.avro.AvroData;
+import org.apache.samza.sql.calcite.data.avro.AvroSchema;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerdeFactory.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerdeFactory.java
similarity index 92%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerdeFactory.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerdeFactory.java
index edd8859..caf4009 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerdeFactory.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerdeFactory.java

@@ -16,14 +16,14 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.impl.data.serializers;
+package org.apache.samza.sql.calcite.data.serializers;
 
 import org.apache.avro.Schema;
 import org.apache.samza.SamzaException;
 import org.apache.samza.config.Config;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.serializers.SerdeFactory;
-import org.apache.samza.operators.impl.data.avro.AvroData;
+import org.apache.samza.sql.calcite.data.avro.AvroData;
 
 public class SqlAvroSerdeFactory implements SerdeFactory<AvroData> {
   public static final String PROP_AVRO_SCHEMA = "serializers.%s.schema";

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlStringSerde.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlStringSerde.java
similarity index 91%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlStringSerde.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlStringSerde.java
index 1267ab6..6651e97 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlStringSerde.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlStringSerde.java

@@ -17,11 +17,11 @@
  * under the License.
  */
 
-package org.apache.samza.operators.impl.data.serializers;
+package org.apache.samza.sql.calcite.data.serializers;
 
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.serializers.StringSerde;
-import org.apache.samza.operators.impl.data.string.StringData;
+import org.apache.samza.sql.calcite.data.string.StringData;
 
 
 public class SqlStringSerde implements Serde<StringData> {

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlStringSerdeFactory.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlStringSerdeFactory.java
similarity index 90%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlStringSerdeFactory.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlStringSerdeFactory.java
index 3b6a3e0..80fb542 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/serializers/SqlStringSerdeFactory.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/serializers/SqlStringSerdeFactory.java

@@ -17,13 +17,13 @@
  * under the License.
  */
 
-package org.apache.samza.operators.impl.data.serializers;
+package org.apache.samza.sql.calcite.data.serializers;
 
 
 import org.apache.samza.config.Config;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.serializers.SerdeFactory;
-import org.apache.samza.operators.impl.data.string.StringData;
+import org.apache.samza.sql.calcite.data.string.StringData;
 
 public class SqlStringSerdeFactory implements SerdeFactory<StringData> {
     @Override

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/string/StringData.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/string/StringData.java
similarity index 94%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/string/StringData.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/string/StringData.java
index 86e9917..f7c8121 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/string/StringData.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/string/StringData.java

@@ -17,10 +17,10 @@
  * under the License.
  */
 
-package org.apache.samza.operators.impl.data.string;
+package org.apache.samza.sql.calcite.data.string;
 
-import org.apache.samza.operators.api.data.Data;
-import org.apache.samza.operators.api.data.Schema;
+import org.apache.samza.sql.calcite.data.Data;
+import org.apache.samza.sql.calcite.data.Schema;
 
 import java.util.List;
 import java.util.Map;

diff --git a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/string/StringSchema.java b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/string/StringSchema.java
similarity index 93%
rename from samza-operator/src/main/java/org/apache/samza/operators/impl/data/string/StringSchema.java
rename to samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/string/StringSchema.java
index b19dfeb..829d61f 100644
--- a/samza-operator/src/main/java/org/apache/samza/operators/impl/data/string/StringSchema.java
+++ b/samza-sql-calcite/src/main/java/org/apache/samza/sql/calcite/data/string/StringSchema.java

@@ -17,10 +17,10 @@
  * under the License.
  */
 
-package org.apache.samza.operators.impl.data.string;
+package org.apache.samza.sql.calcite.data.string;
 
-import org.apache.samza.operators.api.data.Data;
-import org.apache.samza.operators.api.data.Schema;
+import org.apache.samza.sql.calcite.data.Data;
+import org.apache.samza.sql.calcite.data.Schema;
 
 import java.util.Map;
 

diff --git a/samza-operator/src/test/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerdeTest.java b/samza-sql-calcite/src/test/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerdeTest.java
similarity index 85%
rename from samza-operator/src/test/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerdeTest.java
rename to samza-sql-calcite/src/test/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerdeTest.java
index 5aa28bb..b891ddc 100644
--- a/samza-operator/src/test/java/org/apache/samza/operators/impl/data/serializers/SqlAvroSerdeTest.java
+++ b/samza-sql-calcite/src/test/java/org/apache/samza/sql/calcite/data/serializers/SqlAvroSerdeTest.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.impl.data.serializers;
+package org.apache.samza.sql.calcite.data.serializers;
 
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
@@ -27,7 +27,7 @@
 import org.apache.samza.config.Config;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.serializers.Serde;
-import org.apache.samza.operators.impl.data.avro.AvroData;
+import org.apache.samza.sql.calcite.data.avro.AvroData;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -55,10 +55,10 @@
   public void testSqlAvroSerdeDeserialization() throws IOException {
     AvroData decodedDatum = (AvroData)serde.fromBytes(encodeMessage(sampleOrderRecord(), orderSchema));
 
-    Assert.assertTrue(decodedDatum.schema().getType() == org.apache.samza.operators.api.data.Schema.Type.STRUCT);
-    Assert.assertTrue(decodedDatum.getFieldData("id").schema().getType() == org.apache.samza.operators.api.data.Schema.Type.INTEGER);
-    Assert.assertTrue(decodedDatum.getFieldData("quantity").schema().getType() == org.apache.samza.operators.api.data.Schema.Type.INTEGER);
-    Assert.assertTrue(decodedDatum.getFieldData("product").schema().getType() == org.apache.samza.operators.api.data.Schema.Type.STRING);
+    Assert.assertTrue(decodedDatum.schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.STRUCT);
+    Assert.assertTrue(decodedDatum.getFieldData("id").schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.INTEGER);
+    Assert.assertTrue(decodedDatum.getFieldData("quantity").schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.INTEGER);
+    Assert.assertTrue(decodedDatum.getFieldData("product").schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.STRING);
   }
 
   @Test
@@ -69,10 +69,10 @@
 
     AvroData decodedDatum = (AvroData)serde.fromBytes(encodedDatum);
 
-    Assert.assertTrue(decodedDatum.schema().getType() == org.apache.samza.operators.api.data.Schema.Type.STRUCT);
-    Assert.assertTrue(decodedDatum.getFieldData("id").schema().getType() == org.apache.samza.operators.api.data.Schema.Type.INTEGER);
-    Assert.assertTrue(decodedDatum.getFieldData("quantity").schema().getType() == org.apache.samza.operators.api.data.Schema.Type.INTEGER);
-    Assert.assertTrue(decodedDatum.getFieldData("product").schema().getType() == org.apache.samza.operators.api.data.Schema.Type.STRING);
+    Assert.assertTrue(decodedDatum.schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.STRUCT);
+    Assert.assertTrue(decodedDatum.getFieldData("id").schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.INTEGER);
+    Assert.assertTrue(decodedDatum.getFieldData("quantity").schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.INTEGER);
+    Assert.assertTrue(decodedDatum.getFieldData("product").schema().getType() == org.apache.samza.sql.calcite.data.Schema.Type.STRING);
   }
 
   private static Config sqlAvroSerdeTestConfig(){