Move KafkaSchemaRecordType to api package.

Bingqin Zhou · Bingqin Zhou · commit a102e8ccea01 · 2019-12-13T11:17:29.000-08:00
diff --git a/kcbq-api/src/main/java/com/wepay/kafka/connect/bigquery/api/KafkaSchemaRecordType.java b/kcbq-api/src/main/java/com/wepay/kafka/connect/bigquery/api/KafkaSchemaRecordType.java
@@ -1,9 +1,6 @@
-package com.wepay.kafka.connect.bigquery.convert;
+package com.wepay.kafka.connect.bigquery.api;
 
 
-/**
- *  Enum class for Kafka schema or record type, either value or key.
- */
 public enum KafkaSchemaRecordType {
 
     VALUE("value"),
diff --git a/kcbq-api/src/main/java/com/wepay/kafka/connect/bigquery/api/SchemaRetriever.java b/kcbq-api/src/main/java/com/wepay/kafka/connect/bigquery/api/SchemaRetriever.java
@@ -25,7 +25,7 @@ public interface SchemaRetriever {
    * @param schemaType The type of kafka schema, either "value" or "key".
    * @return The Schema for the given table.
    */
-  public Schema retrieveSchema(TableId table, String topic, String schemaType);
+  public Schema retrieveSchema(TableId table, String topic, KafkaSchemaRecordType schemaType);
 
   /**
    * Set the last seen schema for a given topic
diff --git a/kcbq-confluent/src/main/java/com/wepay/kafka/connect/bigquery/schemaregistry/schemaretriever/SchemaRegistrySchemaRetriever.java b/kcbq-confluent/src/main/java/com/wepay/kafka/connect/bigquery/schemaregistry/schemaretriever/SchemaRegistrySchemaRetriever.java
@@ -2,6 +2,7 @@
 
 import com.google.cloud.bigquery.TableId;
 
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;
 
 import io.confluent.connect.avro.AvroData;
@@ -61,7 +62,7 @@ public void configure(Map<String, String> properties) {
   }
 
   @Override
-  public Schema retrieveSchema(TableId table, String topic, String schemaType) {
+  public Schema retrieveSchema(TableId table, String topic, KafkaSchemaRecordType schemaType) {
     String subject = getSubject(topic, schemaType);
     try {
       logger.debug("Retrieving schema information for topic {} with subject {}", topic, subject);
@@ -80,7 +81,7 @@ public Schema retrieveSchema(TableId table, String topic, String schemaType) {
   @Override
   public void setLastSeenSchema(TableId table, String topic, Schema schema) { }
 
-  private String getSubject(String topic, String schemaType) {
-    return topic + "-" + schemaType;
+  private String getSubject(String topic, KafkaSchemaRecordType schemaType) {
+    return topic + "-" + schemaType.toString();
   }
 }
diff --git a/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/BigQuerySinkTask.java b/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/BigQuerySinkTask.java
@@ -25,11 +25,11 @@
 import com.google.cloud.storage.BucketInfo;
 import com.google.cloud.storage.Storage;
 import com.google.common.annotations.VisibleForTesting;
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;
 import com.wepay.kafka.connect.bigquery.config.BigQuerySinkConfig;
 import com.wepay.kafka.connect.bigquery.config.BigQuerySinkTaskConfig;
 import com.wepay.kafka.connect.bigquery.convert.KafkaDataBuilder;
-import com.wepay.kafka.connect.bigquery.convert.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.convert.RecordConverter;
 import com.wepay.kafka.connect.bigquery.convert.SchemaConverter;
 import com.wepay.kafka.connect.bigquery.exception.SinkConfigConnectException;
diff --git a/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/SchemaManager.java b/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/SchemaManager.java
@@ -2,9 +2,9 @@
 
 
 import com.google.cloud.bigquery.*;
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;
 import com.wepay.kafka.connect.bigquery.convert.KafkaDataBuilder;
-import com.wepay.kafka.connect.bigquery.convert.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.convert.SchemaConverter;
 
 import org.apache.kafka.connect.data.Schema;
@@ -52,8 +52,8 @@ public SchemaManager(
    * @param topic The Kafka topic used to determine the schema.
    */
   public void createTable(TableId table, String topic) {
-    Schema kafkaValueSchema = schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.VALUE.toString());
-    Schema kafkaKeySchema = kafkaKeyFieldName.isPresent() ? schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.KEY.toString()) : null;
+    Schema kafkaValueSchema = schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.VALUE);
+    Schema kafkaKeySchema = kafkaKeyFieldName.isPresent() ? schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.KEY) : null;
     bigQuery.create(constructTableInfo(table, kafkaKeySchema, kafkaValueSchema));
   }
 
@@ -63,8 +63,8 @@ public void createTable(TableId table, String topic) {
    * @param topic The Kafka topic used to determine the schema.
    */
   public void updateSchema(TableId table, String topic) {
-    Schema kafkaValueSchema = schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.VALUE.toString());
-    Schema kafkaKeySchema = kafkaKeyFieldName.isPresent() ? schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.KEY.toString()) : null;
+    Schema kafkaValueSchema = schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.VALUE);
+    Schema kafkaKeySchema = kafkaKeyFieldName.isPresent() ? schemaRetriever.retrieveSchema(table, topic, KafkaSchemaRecordType.KEY) : null;
     TableInfo tableInfo = constructTableInfo(table, kafkaKeySchema, kafkaValueSchema);
     logger.info("Attempting to update table `{}` with schema {}",
         table, tableInfo.getDefinition().getSchema());
diff --git a/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/BigQueryRecordConverter.java b/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/BigQueryRecordConverter.java
@@ -19,6 +19,7 @@
 
 
 import com.google.cloud.bigquery.InsertAllRequest.RowToInsert;
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.convert.logicaltype.DebeziumLogicalConverters;
 import com.wepay.kafka.connect.bigquery.convert.logicaltype.KafkaLogicalConverters;
 import com.wepay.kafka.connect.bigquery.convert.logicaltype.LogicalConverterRegistry;
diff --git a/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/RecordConverter.java b/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/RecordConverter.java
@@ -18,6 +18,7 @@
  */
 
 
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import org.apache.kafka.connect.sink.SinkRecord;
 
 /**
diff --git a/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/retrieve/MemorySchemaRetriever.java b/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/retrieve/MemorySchemaRetriever.java
@@ -2,6 +2,7 @@
 
 import com.google.cloud.bigquery.TableId;
 
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;
 
 import org.apache.kafka.common.cache.Cache;
@@ -40,7 +41,7 @@ public void configure(Map<String, String> properties) {
   }
 
   @Override
-  public Schema retrieveSchema(TableId table, String topic, String schemaType) {
+  public Schema retrieveSchema(TableId table, String topic, KafkaSchemaRecordType schemaType) {
     String tableName = table.getTable();
     Schema schema = schemaCache.get(getCacheKey(tableName, topic));
     if (schema != null) {
diff --git a/kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/BigQuerySinkConnectorTest.java b/kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/BigQuerySinkConnectorTest.java
@@ -33,6 +33,7 @@
 import com.google.cloud.bigquery.Table;
 import com.google.cloud.bigquery.TableId;
 
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;
 
 import com.wepay.kafka.connect.bigquery.config.BigQuerySinkConfig;
@@ -61,7 +62,7 @@ public void configure(Map<String, String> properties) {
     }
 
     @Override
-    public Schema retrieveSchema(TableId table, String topic, String subjectType) {
+    public Schema retrieveSchema(TableId table, String topic, KafkaSchemaRecordType schemaType) {
       // Shouldn't be called
       return null;
     }
diff --git a/kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/convert/BigQueryRecordConverterTest.java b/kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/convert/BigQueryRecordConverterTest.java
@@ -20,6 +20,7 @@
 
 import static org.junit.Assert.assertEquals;
 
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.exception.ConversionConnectException;
 
 import org.apache.kafka.connect.data.Schema;
diff --git a/kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/retrieve/MemorySchemaRetrieverTest.java b/kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/retrieve/MemorySchemaRetrieverTest.java
@@ -3,6 +3,7 @@
 
 import com.google.cloud.bigquery.TableId;
 
+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
 import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;
 
 import org.apache.kafka.connect.data.Schema;
@@ -25,7 +26,7 @@ public void testRetrieveSchemaWhenNoLastSeenSchemaReturnsEmptyStructSchema() {
     final TableId tableId = getTableId("testTable", "testDataset");
     SchemaRetriever retriever = new MemorySchemaRetriever();
     retriever.configure(new HashMap<>());
-    Assert.assertEquals(retriever.retrieveSchema(tableId, topic, "value"), SchemaBuilder.struct().build());
+    Assert.assertEquals(retriever.retrieveSchema(tableId, topic, KafkaSchemaRecordType.VALUE), SchemaBuilder.struct().build());
   }
 
   @Test
@@ -38,7 +39,7 @@ public void testRetrieveSchemaWhenLastSeenExistsSucceeds() {
     Schema expectedSchema = Schema.OPTIONAL_FLOAT32_SCHEMA;
     retriever.setLastSeenSchema(tableId, topic, expectedSchema);
 
-    Assert.assertEquals(retriever.retrieveSchema(tableId, topic, "key"), expectedSchema);
+    Assert.assertEquals(retriever.retrieveSchema(tableId, topic, KafkaSchemaRecordType.KEY), expectedSchema);
   }
 
   @Test
@@ -56,8 +57,8 @@ public void testRetrieveSchemaWithMultipleSchemasSucceeds() {
     retriever.setLastSeenSchema(intTableId, intSchemaTopic, expectedIntSchema);
 
     Assert.assertEquals(
-        retriever.retrieveSchema(floatTableId, floatSchemaTopic, "value"), expectedFloatSchema);
-    Assert.assertEquals(retriever.retrieveSchema(intTableId, intSchemaTopic, "key"), expectedIntSchema);
+        retriever.retrieveSchema(floatTableId, floatSchemaTopic, KafkaSchemaRecordType.KEY), expectedFloatSchema);
+    Assert.assertEquals(retriever.retrieveSchema(intTableId, intSchemaTopic, KafkaSchemaRecordType.KEY), expectedIntSchema);
   }
 
   @Test
@@ -72,6 +73,6 @@ public void testRetrieveSchemaRetrievesLastSeenSchema() {
     retriever.setLastSeenSchema(tableId, intSchemaTopic, firstSchema);
     retriever.setLastSeenSchema(tableId, intSchemaTopic, secondSchema);
 
-    Assert.assertEquals(retriever.retrieveSchema(tableId, intSchemaTopic, "value"), secondSchema);
+    Assert.assertEquals(retriever.retrieveSchema(tableId, intSchemaTopic, KafkaSchemaRecordType.VALUE), secondSchema);
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -33,6 +33,7 @@`
`33`	`33`	`import com.google.cloud.bigquery.Table;`
`34`	`34`	`import com.google.cloud.bigquery.TableId;`
`35`	`35`
	`36`	`+import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;`
`36`	`37`	`import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;`
`37`	`38`
`38`	`39`	`import com.wepay.kafka.connect.bigquery.config.BigQuerySinkConfig;`
`@@ -61,7 +62,7 @@ public void configure(Map<String, String> properties) {`
`61`	`62`	`}`
`62`	`63`
`63`	`64`	`@Override`
`64`		`- public Schema retrieveSchema(TableId table, String topic, String subjectType) {`
	`65`	`+ public Schema retrieveSchema(TableId table, String topic, KafkaSchemaRecordType schemaType) {`
`65`	`66`	`// Shouldn't be called`
`66`	`67`	`return null;`
`67`	`68`	`}`