Change kafka data function names.

Author: Bingqin Zhou

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

@@ -157,7 +157,7 @@ private RowToInsert getRecordRow(SinkRecord record) {
       convertedRecord.put(config.getString(config.KAFKA_KEY_FIELD_NAME_CONFIG), recordConverter.convertRecord(record.keySchema(), record.key()));
     }
     if (config.getBoolean(config.INCLUDE_KAFKA_DATA_CONFIG)) {
-      convertedRecord.put(config.getString(config.KAFKA_DATA_FIELD_NAME_CONFIG), KafkaDataBuilder.getKafkaDataRecord(record));
+      convertedRecord.put(config.getString(config.KAFKA_DATA_FIELD_NAME_CONFIG), KafkaDataBuilder.buildKafkaDataRecord(record));
     }
     if (config.getBoolean(config.SANITIZE_FIELD_NAME_CONFIG)) {
       convertedRecord = FieldNameSanitizer.replaceInvalidKeys(convertedRecord);

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

@@ -103,7 +103,7 @@ private com.google.cloud.bigquery.Schema getBigQuerySchema(Schema kafkaKeySchema
           allFields.add(kafkaKeyField);
       }
       if (includeKafkaData) {
-          Field kafkaDataField = KafkaDataBuilder.getKafkaDataField(kafkaDataFieldName);
+          Field kafkaDataField = KafkaDataBuilder.buildKafkaDataField(kafkaDataFieldName);
           allFields.add(kafkaDataField);
       }
       return com.google.cloud.bigquery.Schema.of(allFields);

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

@@ -18,7 +18,7 @@ public class KafkaDataBuilder {
     public static final String KAFKA_DATA_OFFSET_FIELD_NAME = "offset";
     public static final String KAFKA_DATA_INSERT_TIME_FIELD_NAME = "insertTime";
 
-    public static Field getKafkaDataField(String kafkaDataFieldName) {
+    public static Field buildKafkaDataField(String kafkaDataFieldName) {
         Field topicField = com.google.cloud.bigquery.Field.of(KAFKA_DATA_TOPIC_FIELD_NAME, LegacySQLTypeName.STRING);
         Field partitionField = com.google.cloud.bigquery.Field.of(KAFKA_DATA_PARTITION_FIELD_NAME, LegacySQLTypeName.INTEGER);
         Field offsetField = com.google.cloud.bigquery.Field.of(KAFKA_DATA_OFFSET_FIELD_NAME, LegacySQLTypeName.INTEGER);
@@ -32,7 +32,7 @@ public static Field getKafkaDataField(String kafkaDataFieldName) {
                 .setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE).build();
     }
 
-    public static Map<String, Object> getKafkaDataRecord(SinkRecord kafkaConnectRecord) {
+    public static Map<String, Object> buildKafkaDataRecord(SinkRecord kafkaConnectRecord) {
         HashMap<String, Object> kafkaData = new HashMap<>();
         kafkaData.put(KAFKA_DATA_TOPIC_FIELD_NAME, kafkaConnectRecord.topic());
         kafkaData.put(KAFKA_DATA_PARTITION_FIELD_NAME, kafkaConnectRecord.kafkaPartition());

kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/convert/KafkaDataConverterTest.java

@@ -21,7 +21,7 @@ public class KafkaDataConverterTest {
     private static final String kafkaDataInsertTimeName = "insertTime";
 
     @Test
-    public void testGetKafkaDataRecord() {
+    public void testBuildKafkaDataRecord() {
 
         final String kafkaDataTopicValue = "testTopic";
         final int kafkaDataPartitionValue = 101;
@@ -33,7 +33,7 @@ public void testGetKafkaDataRecord() {
         expectedKafkaDataFields.put(kafkaDataOffsetName, kafkaDataOffsetValue);
 
         SinkRecord record = new SinkRecord(kafkaDataTopicValue, kafkaDataPartitionValue, null, null, null, null, kafkaDataOffsetValue);
-        Map<String, Object> actualKafkaDataFields = KafkaDataBuilder.getKafkaDataRecord(record);
+        Map<String, Object> actualKafkaDataFields = KafkaDataBuilder.buildKafkaDataRecord(record);
 
         assertTrue(actualKafkaDataFields.containsKey(kafkaDataInsertTimeName));
         assertTrue(actualKafkaDataFields.get(kafkaDataInsertTimeName) instanceof Double);
@@ -44,7 +44,7 @@ public void testGetKafkaDataRecord() {
     }
 
     @Test
-    public void testGetKafkaDataField() {
+    public void testBuildKafkaDataField() {
         Field topicField = Field.of("topic", LegacySQLTypeName.STRING);
         Field partitionField = Field.of("partition", LegacySQLTypeName.INTEGER);
         Field offsetField = Field.of("offset", LegacySQLTypeName.INTEGER);
@@ -60,7 +60,7 @@ public void testGetKafkaDataField() {
                 insertTimeField)
                 .setMode(Field.Mode.NULLABLE)
                 .build();
-        Field actualBigQuerySchema = KafkaDataBuilder.getKafkaDataField(kafkaDataFieldName);
+        Field actualBigQuerySchema = KafkaDataBuilder.buildKafkaDataField(kafkaDataFieldName);
         assertEquals(expectedBigQuerySchema, actualBigQuerySchema);
     }
 }