Feature/whitelisting

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/BigQueryRecordConverter.java

@@ -80,6 +80,23 @@ public Map<String, Object> convertRecord(SinkRecord kafkaConnectRecord) {
     return convertStruct(kafkaConnectRecord.value(), kafkaConnectSchema);
   }
 
+  /**
+   * Convert a {@link SinkRecord} key into the String Representation
+   *
+   * @param kafkaConnectRecord The Kafka Connect record to convert. Key Must be of type {@link Struct},
+   *                           in order to translate into a row format that requires each field to
+   *                           consist of both a name and a value.
+   * @return The Key values as Map.
+   */
+  public Map<String, Object> convertRecordKey(SinkRecord kafkaConnectRecord) {
+    Schema kafkaConnectSchema = kafkaConnectRecord.keySchema();
+    if (kafkaConnectSchema == null || kafkaConnectSchema.type() != Schema.Type.STRUCT) {
+      throw new
+              ConversionConnectException("Top-level Kafka key schema must be of present and if of type 'struct'");
+    }
+    return convertStruct(kafkaConnectRecord.key(), kafkaConnectSchema);
+  }
+
   @SuppressWarnings("unchecked")
   private Object convertObject(Object kafkaConnectObject, Schema kafkaConnectSchema) {
     if (kafkaConnectObject == null) {

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/RecordConverter.java

@@ -22,12 +22,15 @@
 
 /**
  * Interface for converting from a {@link SinkRecord} to some other kind of record.
+ *
  * @param <R> The type of record to convert to.
  */
 public interface RecordConverter<R> {
-  /**
-   * @param record The record to convert.
-   * @return The converted record.
-   */
-  R convertRecord(SinkRecord record);
+    /**
+     * @param record The record to convert.
+     * @return The converted record.
+     */
+    R convertRecord(SinkRecord record);
+
+    R convertRecordKey(SinkRecord record);
 }

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/kafkadata/KafkaDataBQRecordConverter.java

@@ -63,4 +63,5 @@ public Map<String, Object> convertRecord(SinkRecord kafkaConnectRecord) {
                   System.currentTimeMillis() / 1000.0);
     return record;
   }
+
 }