SQOOP-1930: Sqoop2: Enforce a non empty schema name and column names

(Veena Basavaraj via Abraham Elmahrek)

common/src/main/java/org/apache/sqoop/schema/ByteArraySchema.java

@@ -20,14 +20,16 @@
 import org.apache.sqoop.schema.type.Binary;
 
 /***
- * Schema holding a single field of Binary data
+ * Schema holding a single field of Binary data Used to support connectors to
- * Used to support connectors to schemaless / unstructured systems
+ * schemaless / unstructured systems Such as HDFS or Kafka
- * Such as HDFS or Kafka
  */
 public class ByteArraySchema extends Schema {
 
+  private static final String BYTE_ARRAY_SCHEMA_NAME = "ByteArraySchema";
+  private static final String BYTE_ARRAY_COLUMN_NAME = "ByteArraySchema_Bytes";
+
   public static final ByteArraySchema instance = (ByteArraySchema) new ByteArraySchema()
-          .addColumn(new Binary("ByteArraySchema_Bytes"));
+      .addColumn(new Binary(BYTE_ARRAY_COLUMN_NAME));
 
   public static ByteArraySchema getInstance() {
     return instance;
@@ -35,6 +37,6 @@ public static ByteArraySchema getInstance() {
 
   // To avoid instantiation
   private ByteArraySchema() {
-    super("ByteArraySchema");
+    super(BYTE_ARRAY_SCHEMA_NAME);
   }
 }

common/src/main/java/org/apache/sqoop/schema/NullSchema.java

@@ -19,6 +19,7 @@
 
 public class NullSchema extends Schema {
 
+  private static final String NULL_SCHEMA_NAME = "NullSchema";
   public static final NullSchema instance = new NullSchema();
 
   public static NullSchema getInstance() {
@@ -27,6 +28,6 @@ public static NullSchema getInstance() {
 
   private NullSchema() {
     // empty string is the name
-    super("");
+    super(NULL_SCHEMA_NAME);
   }
 }

common/src/main/java/org/apache/sqoop/schema/Schema.java

@@ -65,7 +65,9 @@ private Schema() {
 
   public Schema(String name) {
     this();
-    assert name != null;
+    if (StringUtils.isEmpty(name)) {
+      throw new SqoopException(SchemaError.SCHEMA_0006, "Schema: " + name);
+    }
     this.name = name;
   }
 
@@ -80,10 +82,6 @@ public Schema(String name) {
    * @return a reference to this object
    */
   public Schema addColumn(Column column) {
-    if(column.getName() == null) {
-      throw new SqoopException(SchemaError.SCHEMA_0001, "Column: " + column);
-    }
-
     if(columNames.contains(column.getName())) {
       throw new SqoopException(SchemaError.SCHEMA_0002, "Column: " + column);
     }

common/src/main/java/org/apache/sqoop/schema/SchemaError.java

@@ -34,7 +34,9 @@ public enum SchemaError implements ErrorCode {
 
   SCHEMA_0004("Non-null target column has no matching source column"),
 
-  SCHEMA_0005("No matching method available for source and target schemas")
+  SCHEMA_0005("No matching method available for source and target schemas"),
+
+  SCHEMA_0006("Schema without name"),
 
   ;
 

common/src/main/java/org/apache/sqoop/schema/type/Column.java

@@ -18,6 +18,8 @@
 package org.apache.sqoop.schema.type;
 
 import org.apache.commons.lang.StringUtils;
+import org.apache.sqoop.common.SqoopException;
+import org.apache.sqoop.schema.SchemaError;
 
 /**
  * Base class for all the supported types in the Sqoop {@link #Schema}
@@ -42,7 +44,9 @@ public Column(String name) {
   }
 
   public Column(String name, Boolean nullable) {
-    assert !StringUtils.isEmpty(name);
+    if (StringUtils.isEmpty(name)) {
+      throw new SqoopException(SchemaError.SCHEMA_0001, "Column name: " + name);
+    }
     this.name = name;
     setNullable(nullable);
   }

common/src/test/java/org/apache/sqoop/json/TestSchemaBean.java

@@ -37,13 +37,10 @@ public class TestSchemaBean extends TestSchemaSerialization {
   protected Schema transfer(Schema schema) {
     SchemaBean extractBean = new SchemaBean(schema);
     JSONObject extractJson = extractBean.extract(true);
-
     String transferredString = extractJson.toJSONString();
-
     JSONObject restoreJson = JSONUtils.parse(transferredString);
     SchemaBean restoreBean = new SchemaBean();
     restoreBean.restore(restoreJson);
-
     return restoreBean.getSchema();
   }
 }

common/src/test/java/org/apache/sqoop/json/util/TestSchemaSerialization.java

@@ -23,7 +23,9 @@
 import java.util.Collections;
 import java.util.HashSet;
 
+import org.apache.sqoop.common.SqoopException;
 import org.apache.sqoop.json.JSONUtils;
+import org.apache.sqoop.schema.ByteArraySchema;
 import org.apache.sqoop.schema.NullSchema;
 import org.apache.sqoop.schema.Schema;
 import org.apache.sqoop.schema.type.Array;
@@ -61,6 +63,35 @@ public void testNullSchemaObject() {
     transferAndAssert(NullSchema.getInstance());
   }
 
+  @SuppressWarnings("unused")
+  @Test(expected = SqoopException.class)
+  public void testEmptySchemaName() {
+    Schema schema = new Schema("");
+  }
+
+  @SuppressWarnings("unused")
+  @Test(expected = SqoopException.class)
+  public void testNullSchemaName() {
+    Schema schema = new Schema(null);
+  }
+
+  @SuppressWarnings("unused")
+  @Test(expected = SqoopException.class)
+  public void testSchemaWithNullColumnName() {
+    Schema schema = new Schema("test").addColumn(new Text(null));
+  }
+
+  @SuppressWarnings("unused")
+  @Test(expected = SqoopException.class)
+  public void testSchemaWithEmptyColumnName() {
+    Schema schema = new Schema("test").addColumn(new Text(""));
+  }
+
+  @Test
+  public void testByteArraySchemaObject() {
+    transferAndAssert(ByteArraySchema.getInstance());
+  }
+
   @Test
   public void testArray() {
     // create an array type containing decimals