// Path: kcbq-api/src/main/java/com/wepay/kafka/connect/bigquery/api/KafkaSchemaRecordType.java
// public enum KafkaSchemaRecordType {
// 
//     VALUE("value"),
//     KEY("key");
// 
//     private final String str;
// 
//     KafkaSchemaRecordType(String str) {
//         this.str = str;
//     }
// 
//     public String toString() {
//         return this.str;
//     }
// }
// 
// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/KafkaDataBuilder.java
// public class KafkaDataBuilder {
// 
//     public static final String KAFKA_DATA_TOPIC_FIELD_NAME = "topic";
//     public static final String KAFKA_DATA_PARTITION_FIELD_NAME = "partition";
//     public static final String KAFKA_DATA_OFFSET_FIELD_NAME = "offset";
//     public static final String KAFKA_DATA_INSERT_TIME_FIELD_NAME = "insertTime";
// 
//     /**
//      * Construct schema for Kafka Data Field
//      *
//      * @param kafkaDataFieldName The configured name of Kafka Data Field
//      * @return Field of Kafka Data, with definitions of kafka topic, partition, offset, and insertTime.
//      */
//     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);
//         Field.Builder insertTimeBuilder = com.google.cloud.bigquery.Field.newBuilder(
//                 KAFKA_DATA_INSERT_TIME_FIELD_NAME, LegacySQLTypeName.TIMESTAMP)
//                 .setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE);
//         Field insertTimeField = insertTimeBuilder.build();
// 
//         return Field.newBuilder(kafkaDataFieldName, LegacySQLTypeName.RECORD,
//                 topicField, partitionField, offsetField, insertTimeField)
//                 .setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE).build();
//     }
// 
//     /**
//      * Construct a map of Kafka Data record
//      *
//      * @param kafkaConnectRecord Kafka sink record to build kafka data from.
//      * @return HashMap which contains the values of kafka topic, partition, offset, and insertTime.
//      */
//     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());
//         kafkaData.put(KAFKA_DATA_OFFSET_FIELD_NAME, kafkaConnectRecord.kafkaOffset());
//         kafkaData.put(KAFKA_DATA_INSERT_TIME_FIELD_NAME, System.currentTimeMillis() / 1000.0);
//         return kafkaData;
//     }
// 
// }
// 
// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/RecordConverter.java
// public interface RecordConverter<R> {
//   /**
//    * @param record The record to convert.
//    * @param recordType The type of the record to convert, either value or key.
//    * @return The converted record.
//    */
//   R convertRecord(SinkRecord record, KafkaSchemaRecordType recordType);
// 
// }


import com.google.cloud.bigquery.InsertAllRequest;
import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
import com.wepay.kafka.connect.bigquery.convert.KafkaDataBuilder;
import com.wepay.kafka.connect.bigquery.convert.RecordConverter;
import org.apache.kafka.connect.sink.SinkRecord;
import java.util.Map;
import java.util.Optional;
package com.wepay.kafka.connect.bigquery.utils;

/*
 * Copyright 2016 WePay, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */



/**
 * A class for converting a {@link SinkRecord SinkRecord} to {@link InsertAllRequest.RowToInsert BigQuery row}
 */
public class SinkRecordConverter {
    private final RecordConverter<Map<String, Object>> recordConverter;
    private final boolean sanitizeFieldName;
    private final Optional<String> kafkaKeyFieldName;
    private final Optional<String> kafkaDataFieldName;

    public SinkRecordConverter(RecordConverter<Map<String, Object>> recordConverter, boolean sanitizeFieldName, Optional<String> kafkaKeyFieldName, Optional<String> kafkaDataFieldName) {
        this.recordConverter = recordConverter;
        this.sanitizeFieldName = sanitizeFieldName;
        this.kafkaKeyFieldName = kafkaKeyFieldName;
        this.kafkaDataFieldName = kafkaDataFieldName;
    }

    public InsertAllRequest.RowToInsert getRecordRow(SinkRecord record) {
        Map<String, Object> convertedRecord = recordConverter.convertRecord(record, KafkaSchemaRecordType.VALUE);
        if (kafkaKeyFieldName.isPresent()) {
            convertedRecord.put(kafkaKeyFieldName.get(), recordConverter.convertRecord(record, KafkaSchemaRecordType.KEY));
        }
        if (kafkaDataFieldName.isPresent()) {
// Path: kcbq-api/src/main/java/com/wepay/kafka/connect/bigquery/api/KafkaSchemaRecordType.java
// public enum KafkaSchemaRecordType {
// 
//     VALUE("value"),
//     KEY("key");
// 
//     private final String str;
// 
//     KafkaSchemaRecordType(String str) {
//         this.str = str;
//     }
// 
//     public String toString() {
//         return this.str;
//     }
// }
// 
// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/KafkaDataBuilder.java
// public class KafkaDataBuilder {
// 
//     public static final String KAFKA_DATA_TOPIC_FIELD_NAME = "topic";
//     public static final String KAFKA_DATA_PARTITION_FIELD_NAME = "partition";
//     public static final String KAFKA_DATA_OFFSET_FIELD_NAME = "offset";
//     public static final String KAFKA_DATA_INSERT_TIME_FIELD_NAME = "insertTime";
// 
//     /**
//      * Construct schema for Kafka Data Field
//      *
//      * @param kafkaDataFieldName The configured name of Kafka Data Field
//      * @return Field of Kafka Data, with definitions of kafka topic, partition, offset, and insertTime.
//      */
//     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);
//         Field.Builder insertTimeBuilder = com.google.cloud.bigquery.Field.newBuilder(
//                 KAFKA_DATA_INSERT_TIME_FIELD_NAME, LegacySQLTypeName.TIMESTAMP)
//                 .setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE);
//         Field insertTimeField = insertTimeBuilder.build();
// 
//         return Field.newBuilder(kafkaDataFieldName, LegacySQLTypeName.RECORD,
//                 topicField, partitionField, offsetField, insertTimeField)
//                 .setMode(com.google.cloud.bigquery.Field.Mode.NULLABLE).build();
//     }
// 
//     /**
//      * Construct a map of Kafka Data record
//      *
//      * @param kafkaConnectRecord Kafka sink record to build kafka data from.
//      * @return HashMap which contains the values of kafka topic, partition, offset, and insertTime.
//      */
//     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());
//         kafkaData.put(KAFKA_DATA_OFFSET_FIELD_NAME, kafkaConnectRecord.kafkaOffset());
//         kafkaData.put(KAFKA_DATA_INSERT_TIME_FIELD_NAME, System.currentTimeMillis() / 1000.0);
//         return kafkaData;
//     }
// 
// }
// 
// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/RecordConverter.java
// public interface RecordConverter<R> {
//   /**
//    * @param record The record to convert.
//    * @param recordType The type of the record to convert, either value or key.
//    * @return The converted record.
//    */
//   R convertRecord(SinkRecord record, KafkaSchemaRecordType recordType);
// 
// }


// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/utils/SinkRecordConverter.java
import com.google.cloud.bigquery.InsertAllRequest;
import com.wepay.kafka.connect.bigquery.api.KafkaSchemaRecordType;
import com.wepay.kafka.connect.bigquery.convert.KafkaDataBuilder;
import com.wepay.kafka.connect.bigquery.convert.RecordConverter;
import org.apache.kafka.connect.sink.SinkRecord;
import java.util.Map;
import java.util.Optional;

package com.wepay.kafka.connect.bigquery.utils;

/*
 * Copyright 2016 WePay, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */



/**
 * A class for converting a {@link SinkRecord SinkRecord} to {@link InsertAllRequest.RowToInsert BigQuery row}
 */
public class SinkRecordConverter {
    private final RecordConverter<Map<String, Object>> recordConverter;
    private final boolean sanitizeFieldName;
    private final Optional<String> kafkaKeyFieldName;
    private final Optional<String> kafkaDataFieldName;

    public SinkRecordConverter(RecordConverter<Map<String, Object>> recordConverter, boolean sanitizeFieldName, Optional<String> kafkaKeyFieldName, Optional<String> kafkaDataFieldName) {
        this.recordConverter = recordConverter;
        this.sanitizeFieldName = sanitizeFieldName;
        this.kafkaKeyFieldName = kafkaKeyFieldName;
        this.kafkaDataFieldName = kafkaDataFieldName;
    }

    public InsertAllRequest.RowToInsert getRecordRow(SinkRecord record) {
        Map<String, Object> convertedRecord = recordConverter.convertRecord(record, KafkaSchemaRecordType.VALUE);
        if (kafkaKeyFieldName.isPresent()) {
            convertedRecord.put(kafkaKeyFieldName.get(), recordConverter.convertRecord(record, KafkaSchemaRecordType.KEY));
        }
        if (kafkaDataFieldName.isPresent()) {
            convertedRecord.put(kafkaDataFieldName.get(), KafkaDataBuilder.buildKafkaDataRecord(record));