content: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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.
   */
  package org.apache.apex.malhar.kudu;
  
  import java.io.Serializable;
  import java.util.ArrayList;
  import java.util.List;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.kudu.client.ExternalConsistencyMode;
  import org.apache.kudu.client.KuduClient;
  import org.apache.kudu.client.KuduSession;
  import org.apache.kudu.client.KuduTable;
  import org.apache.kudu.client.SessionConfiguration;
  
  import static com.google.common.base.Preconditions.checkArgument;
  import static com.google.common.base.Preconditions.checkNotNull;
  
  /**
   * <p>
   * Represents a connection to the Kudu cluster. An instance of this class is to
   * be supplied (via a builder pattern to) {@link AbstractKuduOutputOperator} to
   * connect to a Kudu cluster.
   * </p>
   *
   * @since 3.8.0
   */
  
  public class ApexKuduConnection implements AutoCloseable, Serializable
  {
    private static final long serialVersionUID = 4720185362997969198L;
  
    private transient  KuduSession kuduSession;
  
    private transient  KuduTable kuduTable;
  
    private transient  KuduClient kuduClient;
  
    public static final Logger LOG = LoggerFactory.getLogger(ApexKuduConnection.class);
  
    private ApexKuduConnectionBuilder builderForThisConnection;
  
    private ApexKuduConnection(ApexKuduConnectionBuilder builder)
    {
      checkNotNull(builder,"Builder cannot be null to establish kudu session");
      checkArgument(builder.mastersCollection.size() > 0, "Atleast one kudu master needs to be specified");
      checkNotNull(builder.tableName,"Kudu table cannot be null");
      builderForThisConnection = builder;
      KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(builder.mastersCollection);
      if (builder.isOperationTimeOutSet) {
        kuduClientBuilder.defaultOperationTimeoutMs(builder.operationTimeOutMs);
      }
      if (builder.isBossThreadCountSet) {
        kuduClientBuilder.bossCount(builder.numBossThreads);
      }
      if (builder.isWorkerThreadsCountSet) {
        kuduClientBuilder.workerCount(builder.workerThreads);
      }
      if (builder.isSocketReadTimeOutSet) {
        kuduClientBuilder.defaultSocketReadTimeoutMs(builder.socketReadTimeOutMs);
      }
      kuduClient = kuduClientBuilder.build();
      kuduSession = kuduClient.newSession();
      if (builder.isFlushModeSet) {
        kuduSession.setFlushMode(builder.flushMode);
      }
      if (builder.isExternalConsistencyModeSet) {
        kuduSession.setExternalConsistencyMode(builder.externalConsistencyMode);
      }
      try {
        if (!kuduClient.tableExists(builder.tableName)) {
          throw new Exception("Table " + builder.tableName + " does not exist. ");
        } else {
          kuduTable = kuduClient.openTable(builder.tableName);
        }
      } catch (Exception e) {
        throw new RuntimeException("Kudu table existence could not be ascertained  " + e.getMessage(), e);
      }
    }
  
    @Override
    public void close() throws Exception
    {
      kuduSession.close();
      kuduClient.close();
    }
  
    public KuduSession getKuduSession()
    {
      return kuduSession;
    }
  
    public void setKuduSession(KuduSession kuduSession)
    {
      this.kuduSession = kuduSession;
    }
  
    public KuduTable getKuduTable()
    {
      return kuduTable;
    }
  
    public void setKuduTable(KuduTable kuduTable)
    {
      this.kuduTable = kuduTable;
    }
  
    public KuduClient getKuduClient()
    {
      return kuduClient;
    }
  
    public void setKuduClient(KuduClient kuduClient)
    {
      this.kuduClient = kuduClient;
    }
  
    public ApexKuduConnectionBuilder getBuilderForThisConnection()
    {
      return builderForThisConnection;
    }
  
    public void setBuilderForThisConnection(ApexKuduConnectionBuilder builderForThisConnection)
    {
      this.builderForThisConnection = builderForThisConnection;
    }
  
    public static class ApexKuduConnectionBuilder implements Serializable
    {
      private static final long serialVersionUID = -3428649955056723311L;
  
      List<String> mastersCollection = new ArrayList<>();
  
      String tableName;
  
      // optional props
      int numBossThreads = 1;
  
      boolean isBossThreadCountSet = false;
  
      int workerThreads = 2 *  Runtime.getRuntime().availableProcessors();
  
      boolean isWorkerThreadsCountSet = false;
  
      long socketReadTimeOutMs = 10000;
  
      boolean isSocketReadTimeOutSet = false;
  
      long operationTimeOutMs = 30000;
  
      boolean isOperationTimeOutSet = false;
  
      ExternalConsistencyMode externalConsistencyMode;
  
      boolean isExternalConsistencyModeSet = false;
  
      SessionConfiguration.FlushMode flushMode;
  
      boolean isFlushModeSet = false;
  
      public ApexKuduConnectionBuilder withTableName(String tableName)
      {
        this.tableName = tableName;
        return this;
      }
  
      public ApexKuduConnectionBuilder withAPossibleMasterHostAs(String masterHostAndPort)
      {
        mastersCollection.add(masterHostAndPort);
        return this;
      }
  
      public ApexKuduConnectionBuilder withNumberOfBossThreads(int numberOfBossThreads)
      {
        this.numBossThreads = numberOfBossThreads;
        isBossThreadCountSet = true;
        return this;
      }
  
      public ApexKuduConnectionBuilder withNumberOfWorkerThreads(int numberOfWorkerThreads)
      {
        this.workerThreads = numberOfWorkerThreads;
        isWorkerThreadsCountSet = true;
        return this;
      }
  
      public ApexKuduConnectionBuilder withSocketReadTimeOutAs(long socketReadTimeOut)
      {
        this.socketReadTimeOutMs = socketReadTimeOut;
        isSocketReadTimeOutSet = true;
        return this;
      }
  
      public ApexKuduConnectionBuilder withOperationTimeOutAs(long operationTimeOut)
      {
        this.operationTimeOutMs = operationTimeOut;
        isOperationTimeOutSet = true;
        return this;
      }
  
      public ApexKuduConnectionBuilder withExternalConsistencyMode(ExternalConsistencyMode externalConsistencyMode)
      {
        this.externalConsistencyMode = externalConsistencyMode;
        isExternalConsistencyModeSet = true;
        return this;
      }
  
      public ApexKuduConnectionBuilder withFlushMode(SessionConfiguration.FlushMode flushMode)
      {
        this.flushMode = flushMode;
        isFlushModeSet = true;
        return this;
      }
  
      public ApexKuduConnection build()
      {
        ApexKuduConnection apexKuduConnection = new ApexKuduConnection(this);
        return apexKuduConnection;
      }
  
    }
  }
  