// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/exception/BigQueryConnectException.java
// public class BigQueryConnectException extends ConnectException {
//   public BigQueryConnectException(String msg) {
//     super(msg);
//   }
// 
//   public BigQueryConnectException(String msg, Throwable thr) {
//     super(msg, thr);
//   }
// 
//   public BigQueryConnectException(Throwable thr) {
//     super(thr);
//   }
// 
//   public BigQueryConnectException(Map<Long, List<BigQueryError>> errors) {
//     super(formatInsertAllErrors(errors));
//   }
// 
//   private static String formatInsertAllErrors(Map<Long, List<BigQueryError>> errorsMap) {
//     StringBuilder messageBuilder = new StringBuilder();
//     messageBuilder.append("table insertion failed for the following rows:");
//     for (Map.Entry<Long, List<BigQueryError>> errorsEntry : errorsMap.entrySet()) {
//       for (BigQueryError error : errorsEntry.getValue()) {
//         messageBuilder.append(String.format(
//             "%n\t[row index %d]: %s: %s",
//             errorsEntry.getKey(),
//             error.getReason(),
//             error.getMessage()
//         ));
//       }
//     }
//     return messageBuilder.toString();
//   }
// }


import java.nio.charset.StandardCharsets;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.wepay.kafka.connect.bigquery.exception.BigQueryConnectException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ByteArrayInputStream;
package com.wepay.kafka.connect.bigquery;

/*
 * 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.
 */





/**
 * Convenience class for creating a default {@link com.google.cloud.bigquery.BigQuery} instance,
 * with or without login credentials.
 */
public class BigQueryHelper {
  private static final Logger logger = LoggerFactory.getLogger(BigQueryHelper.class);
  private static String keySource;

  /**
   * Returns a default {@link BigQuery} instance for the specified project with credentials provided
   * in the specified file, which can then be used for creating, updating, and inserting into tables
   * from specific datasets.
   *
   * @param projectName The name of the BigQuery project to work with
   * @param key The google credentials JSON key that can be used to provide
   *                    credentials to BigQuery, or null if no authentication should be performed.
   * @return The resulting BigQuery object.
   */
  public BigQuery connect(String projectName, String key) {
    if (key == null) {
      return connect(projectName);
    }
    logger.debug("Attempting to open file {} for service account json key", key);
    InputStream credentialsStream;
    try {
      if (keySource != null && keySource.equals("JSON")) {
        credentialsStream = new ByteArrayInputStream(key.getBytes(StandardCharsets.UTF_8));
      } else {
        credentialsStream = new FileInputStream(key);
      }
      return new
          BigQueryOptions.DefaultBigQueryFactory().create(
              BigQueryOptions.newBuilder()
                  .setProjectId(projectName)
                  .setCredentials(GoogleCredentials.fromStream(credentialsStream))
                  .build()
      );
    } catch (IOException err) {
// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/exception/BigQueryConnectException.java
// public class BigQueryConnectException extends ConnectException {
//   public BigQueryConnectException(String msg) {
//     super(msg);
//   }
// 
//   public BigQueryConnectException(String msg, Throwable thr) {
//     super(msg, thr);
//   }
// 
//   public BigQueryConnectException(Throwable thr) {
//     super(thr);
//   }
// 
//   public BigQueryConnectException(Map<Long, List<BigQueryError>> errors) {
//     super(formatInsertAllErrors(errors));
//   }
// 
//   private static String formatInsertAllErrors(Map<Long, List<BigQueryError>> errorsMap) {
//     StringBuilder messageBuilder = new StringBuilder();
//     messageBuilder.append("table insertion failed for the following rows:");
//     for (Map.Entry<Long, List<BigQueryError>> errorsEntry : errorsMap.entrySet()) {
//       for (BigQueryError error : errorsEntry.getValue()) {
//         messageBuilder.append(String.format(
//             "%n\t[row index %d]: %s: %s",
//             errorsEntry.getKey(),
//             error.getReason(),
//             error.getMessage()
//         ));
//       }
//     }
//     return messageBuilder.toString();
//   }
// }


// Path: kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/BigQueryHelper.java
import java.nio.charset.StandardCharsets;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.wepay.kafka.connect.bigquery.exception.BigQueryConnectException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ByteArrayInputStream;

package com.wepay.kafka.connect.bigquery;

/*
 * 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.
 */





/**
 * Convenience class for creating a default {@link com.google.cloud.bigquery.BigQuery} instance,
 * with or without login credentials.
 */
public class BigQueryHelper {
  private static final Logger logger = LoggerFactory.getLogger(BigQueryHelper.class);
  private static String keySource;

  /**
   * Returns a default {@link BigQuery} instance for the specified project with credentials provided
   * in the specified file, which can then be used for creating, updating, and inserting into tables
   * from specific datasets.
   *
   * @param projectName The name of the BigQuery project to work with
   * @param key The google credentials JSON key that can be used to provide
   *                    credentials to BigQuery, or null if no authentication should be performed.
   * @return The resulting BigQuery object.
   */
  public BigQuery connect(String projectName, String key) {
    if (key == null) {
      return connect(projectName);
    }
    logger.debug("Attempting to open file {} for service account json key", key);
    InputStream credentialsStream;
    try {
      if (keySource != null && keySource.equals("JSON")) {
        credentialsStream = new ByteArrayInputStream(key.getBytes(StandardCharsets.UTF_8));
      } else {
        credentialsStream = new FileInputStream(key);
      }
      return new
          BigQueryOptions.DefaultBigQueryFactory().create(
              BigQueryOptions.newBuilder()
                  .setProjectId(projectName)
                  .setCredentials(GoogleCredentials.fromStream(credentialsStream))
                  .build()
      );
    } catch (IOException err) {
      throw new BigQueryConnectException("Failed to access json key file", err);