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.IOException;
  import java.lang.reflect.Field;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  
  import javax.validation.constraints.NotNull;
  
  import org.codehaus.jackson.annotate.JsonIgnore;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.api.ControlAwareDefaultOutputPort;
  import org.apache.apex.malhar.kudu.partitioner.AbstractKuduInputPartitioner;
  import org.apache.apex.malhar.kudu.partitioner.KuduOneToManyPartitioner;
  import org.apache.apex.malhar.kudu.partitioner.KuduOneToOnePartitioner;
  import org.apache.apex.malhar.kudu.partitioner.KuduPartitionScanStrategy;
  import org.apache.apex.malhar.kudu.scanner.AbstractKuduPartitionScanner;
  import org.apache.apex.malhar.kudu.scanner.KuduPartitionConsistentOrderScanner;
  import org.apache.apex.malhar.kudu.scanner.KuduPartitionRandomOrderScanner;
  import org.apache.apex.malhar.kudu.scanner.KuduPartitionScanAssignmentMeta;
  import org.apache.apex.malhar.kudu.scanner.KuduRecordWithMeta;
  import org.apache.apex.malhar.kudu.scanner.KuduScanOrderStrategy;
  import org.apache.apex.malhar.kudu.sqltranslator.KuduSQLParseTreeListener;
  import org.apache.apex.malhar.kudu.sqltranslator.SQLToKuduPredicatesTranslator;
  import org.apache.apex.malhar.lib.util.PojoUtils;
  import org.apache.apex.malhar.lib.wal.FSWindowDataManager;
  import org.apache.apex.malhar.lib.wal.WindowDataManager;
  import org.apache.hadoop.classification.InterfaceStability;
  import org.apache.kudu.ColumnSchema;
  import org.apache.kudu.Common;
  import org.apache.kudu.client.KuduTable;
  
  import com.conversantmedia.util.concurrent.DisruptorBlockingQueue;
  import com.conversantmedia.util.concurrent.SpinPolicy;
  
  import com.datatorrent.api.Context;
  import com.datatorrent.api.DefaultOutputPort;
  import com.datatorrent.api.InputOperator;
  import com.datatorrent.api.Operator;
  import com.datatorrent.api.Partitioner;
  import com.datatorrent.api.StatsListener;
  import com.datatorrent.api.annotation.Stateless;
  
  import static com.google.common.base.Preconditions.checkNotNull;
  
  /**
   * <p>Abstract Kudu Input Operator that can be used to stream POJO representation of a Kudu table row by providing a
   *  SQL expression as an input to the operator. The SQL expression is fed into the operator by implementing the
   *   {@link AbstractKuduInputOperator#getNextQuery()}. The query processing is a continuous process wherein the
   *    method is invoked as soon as the operator is done processing the current query</p>
   *
   * <p>
   *   The following are the main features of  the Kudu input operator
   *   <ol>
   *     <li>The scans are performed using a SQL expression. The SQL expression is limited to the constructs
   *      provided by the Kudu scan engine. As an example not equal to expressions are not allowed as predicates</li>
   *     <li>The SQL expression allows for scanning with options that allow for setting the read snap shot time
   *      and/or setting the message that will be sent as part of the control tuple after the query is sent to the
   *      downstream operators</li>
   *     <li>The operator allows for a custom control tuple to be sent when the query data is streamed to the downstream
   *      operators</li>
   *     <li>The operator is templatised.
   *       <ol>
   *         <li>The first template variable represents the POJO class that represents the single
   *      row of a kudu table</li>
   *        <li>The second template parameter represents the control tuple message that is sent to downstream operators
   *         once a query is complete.</li>
   *       </ol>
   *     </li>
   *     <li>
   *       The input operator can be configured as a fault tolerant input operator. Configuring for fault tolerance
   *       implies that the scan is slower but resilient for tablet server failures. See
   *       {@link AbstractKuduInputOperator#setFaultTolerantScanner(boolean)}</li>
   *     </li>
   *     <li>The number of partitions can be set via typical apex configuration or via the method
   *     {@link AbstractKuduInputOperator#setNumberOfPartitions(int)}. The configuration value is overridden by the
   *     API method if both are defined.</li>
   *     <li>
   *       A partition scan strategy can be configured. There are two types os supported scan partitioners.Default is
   *       MANY_TABLETS_PER_OPERATOR.
   *       <ol>
   *         <li>MANY_TABLETS_PER_OPERATOR: Many kudu tablets are mapped to one Apex parition while scanning. See
   *          {@link KuduOneToManyPartitioner}</li>
   *         <li>ONE_TABLET_PER_OPERATOR: One Kudu tablet to one Apex partition. See
   *          {@link KuduOneToOnePartitioner}</li>
   *       </ol>
   *     </li>
   *     <li>A scan order strategy can also be set. This configuration allows for two types of ordering. The default
   *     is RANDOM_ORDER_SCANNER.
   *      <ol>
   *        <li>CONSISTENT_ORDER_SCANNER: Ensures the scans are consistent across checkpoint and restarts. See
   *         {@link KuduPartitionConsistentOrderScanner}</li>
   *        <li>RANDOM_ORDER_SCANNER: Rows from different kudu tablets are streamed in parallel in different threads.
   *         Choose this option for the highest throughput but with the downside of no ordering guarantees across
   *          multiple launches of the operator. See {@link KuduPartitionRandomOrderScanner}</li>
   *      </ol>
   *     </li>
   *     <li>Reads from Kudu tablets happen on a different thread than the main operator thread. The operator uses the
   *      DisruptorBlockingQueue as the buffer to achieve optimal performance and high throughput. The data scanned is
   *      sent via this buffer to the kudu input operator. Depending on the configuration, there can be multiple threads
   *       scanning multiple kudu tablets in parallel. The input operator allows to fine tune the buffering strategy.
   *        Set the CPU policy to SPINNING to
   *     get the highest performance. However this setting results in showing a 100% CPU busy state on the host where
   *      it is spinning instead of doing any useful work until there is data processing required</li>
   *     <li>The Operator allows for exactly once semantics when resuming from a checkpoint after a crash recovery/
   *     restart. For the exactly once semantics, the operator developer who is extending the Abstract input
   *      operator will need to override the method
   *      {@link AbstractKuduInputOperator#isAllowedInReconcilingWindow(KuduRecordWithMeta)}. Return true if the
   *      row needs to be streamed downstream. Note that exactly once semantics requires a lot more
   *       complexity in a truly distributed framework. Note that this method is called only for one window when
   *        the operator is in a reconciling mode in the last window that it was operating in just before a crash
   *         or a shutdown of the application.</li>
   *     <li>The query processing is done by operator one query after another and is independent of the checkpoint
   *      boundary. This essentially means that any given instance of time all physical instances of the operator
   *       are not processing the same query. Downstream operators if using a unifier need to be aware of this. The
   *        control tuple sent downstream can be used to identify the boundaries of the operator query processing
   *         boundaries.
   *     {@link AbstractKuduInputOperator#getNextQuery()} method is used by the child classes of this operator to
   *      define the next SQL expression that needs to be processed.</li>
   *     <li>Connection to the Kudu cluster is managed by specifying a
   *      {@link ApexKuduConnection.ApexKuduConnectionBuilder)}. Use this builder to fine tune various connection configs
   *       to ensure that the optimal resources are set for the scanner</li>
   *     <li>The tuple throughput can be controlled by using the
   *      {@link AbstractKuduInputOperator#setMaxTuplesPerWindow(int)} to control the rate at which the
   *       tuples are emitted in any given window</li>
   *     <li>Note that the default window data manager is the FS based window data manger. Changing this to Noop or
   *      other dummy implementations would result in 'EXACTLY ONCE' semantics void.</li>
   *   </ol>
   * </p>
   * @since 3.8.0
   */
  @InterfaceStability.Evolving
  public abstract class AbstractKuduInputOperator<T,C extends InputOperatorControlTuple> implements InputOperator,
      Operator.ActivationListener<Context.OperatorContext>, Operator.CheckpointNotificationListener,
      Partitioner<AbstractKuduInputOperator>, StatsListener
  {
  
    private static final Logger LOG = LoggerFactory.getLogger(AbstractKuduInputOperator.class);
  
    private KuduPartitionScanStrategy partitionScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR;
  
    private KuduScanOrderStrategy scanOrderStrategy = KuduScanOrderStrategy.RANDOM_ORDER_SCANNER;
  
    protected ApexKuduConnection.ApexKuduConnectionBuilder apexKuduConnectionInfo;
  
    /** Represents the PIE of the total tablets that are distributed across all instances of the operator. Note that
     *  this <b>DOES NOT</b> represent the work distribution for a given query for reasons related to churn of
     *   partitioning. Any query that is processed is assigned is alloted a subset of this pie with the possibility of
     *   all the query utilizing all of the pie for computing the query.
    */
    private List<KuduPartitionScanAssignmentMeta> partitionPieAssignment = new ArrayList<>();
  
    protected Class<T> clazzForResultObject;
  
    private int numberOfPartitions = -1; // set to -1 to see in partition logic to see if user explicitly set this value
  
    protected String tableName;
  
    // performance fine tuning params
  
    /**
     *  Set this to false to increase throughput but at the risk of being Kudu tablet failure dependent
     */
    private boolean isFaultTolerantScanner = true;
  
    private SpinPolicy cpuSpinPolicyForWaitingInBuffer = SpinPolicy.WAITING;
  
    private int bufferCapacity = 8192;
  
    private int maxTuplesPerWindow = -1;
  
  
    // current query fields
    private String currentQueryBeingProcessed;
  
    private Map<String,String> optionsEnabledForCurrentQuery;
  
    /**
     * Used to track the number of scantokens that are valid for this operator instance and this query. Used
     * to track the current query processing status and whether the operator is ready to accept the next query.
     */
    private int plannedSegmentsForCurrentQuery = 0;
  
    /**
     * Used to track the completion status of each scantoken Kudu partition that was assigned for this query and this
     * physical instance of the operator.
     */
    private Map<KuduPartitionScanAssignmentMeta,Boolean> currentQueryCompletionStatus = new HashMap<>();
  
    private boolean allScansCompleteForThisQuery = true;
  
    private boolean isPartitioned = false;
  
    @NotNull
    private WindowDataManager windowDataManager = new FSWindowDataManager();
  
    private transient long currentWindowId;
  
    private transient long reconcilingPhaseWindowId;
  
    private transient boolean isCurrentlyInSafeMode;
  
    private transient boolean isCurrentlyInReconcilingMode;
  
    private transient Map<KuduPartitionScanAssignmentMeta,Long> windowManagerDataForScans;
  
    @JsonIgnore
    protected transient AbstractKuduInputPartitioner partitioner;
  
    @JsonIgnore
    protected transient AbstractKuduPartitionScanner<T,C> scanner;
  
    private transient DisruptorBlockingQueue<KuduRecordWithMeta<T>> buffer;
  
    private transient Map<String,String> kuduColNameToPOJOFieldNameMap;
  
    private transient  Map<String,ColumnSchema> kuduColNameToSchemaMapping;
  
    private transient int currentWindowTupleCount = 0;
  
    public final transient ControlAwareDefaultOutputPort<T> outputPort = new ControlAwareDefaultOutputPort<>();
  
    public final transient  DefaultOutputPort<String> errorPort = new DefaultOutputPort<>();
  
    public AbstractKuduInputOperator()
    {
      // Used by the Kryo cloning process.
      // Other usages would need to set the table name, class and Kudu connection using setters.
      // Kryo would take care of these values as it does a deep copy.
    }
  
    public AbstractKuduInputOperator(ApexKuduConnection.ApexKuduConnectionBuilder kuduConnectionInfo,
        Class<T> clazzForPOJO) throws Exception
    {
      checkNotNull(clazzForPOJO," Class definition for POJO cannot be null");
      checkNotNull(kuduConnectionInfo, "Kudu connection info cannot be null");
      apexKuduConnectionInfo = kuduConnectionInfo;
      clazzForResultObject = clazzForPOJO;
      tableName = kuduConnectionInfo.tableName;
    }
  
    /**
     * Implement this method to get the next query that needs to be processed by this instance of the operator. Note
     * that the given query is distributed across all physical instances of the operator. It is entirely possible that
     * each physical instance of the operator is processing a different query at any given instance of time because
     * the rows that need to be scanned might not be equally spread across all kudu partitions.
     * As long as all instances of the physical operator get the same sequence of queries as part of the implementation
     *  of this method, all the physical operator instances will ensure the same sequence of processing of the queries.
     * @return The next query string that needs to be processed.
     */
    protected abstract String getNextQuery();
  
  
    /**
    * Implements  the logic to process a given query. Note that this method is invoked from two states.
     *  One state when the Operator is resuming from a checkpointed state and the second when a new query needs to be
     *   processed as part of the normal execution processing.
    */
    protected boolean processForQueryString(String queryExpression)
    {
      LOG.info("Processing query " + queryExpression);
      SQLToKuduPredicatesTranslator parsedQuery = null;
      try {
        parsedQuery = new SQLToKuduPredicatesTranslator(
          queryExpression,new ArrayList<ColumnSchema>(kuduColNameToSchemaMapping.values()));
        parsedQuery.parseKuduExpression();
      } catch (Exception e) {
        LOG.error("Could not parse the SQL expression/query " + e.getMessage(),e);
        errorPort.emit(queryExpression);
        return false;
      }
      if (parsedQuery.getErrorListener().isSyntaxError()) {
        LOG.error(" Query is an invalid query " + queryExpression);
        errorPort.emit(queryExpression);
        return false;
      }
      if (!parsedQuery.getKuduSQLParseTreeListener().isSuccessfullyParsed()) {
        LOG.error(" Query could not be successfully parsed instead of being syntactically correct " + queryExpression);
        errorPort.emit(queryExpression);
        return false;
      }
      Map<String,Object> setters = extractSettersForResultObject(parsedQuery);
      try {
        currentQueryBeingProcessed = queryExpression;
        optionsEnabledForCurrentQuery.clear();
        optionsEnabledForCurrentQuery.putAll(parsedQuery.getKuduSQLParseTreeListener().getOptionsUsed());
        plannedSegmentsForCurrentQuery = scanner.scanAllRecords(parsedQuery,setters);
      } catch (IOException e) {
        LOG.error("Error while scanning the kudu segments " + e.getMessage(), e);
        errorPort.emit(queryExpression);
        return false;
      }
      LOG.info("Query" + queryExpression + " submitted for scanning");
      return true;
    }
  
    /**
    * Resets the processing state of data structures that are applicable at a query level.
    */
    protected boolean processForNextQuery()
    {
      LOG.info("Clearing data structures for processing query " + currentQueryBeingProcessed);
      windowManagerDataForScans.clear();
      currentQueryCompletionStatus.clear();
      return processForQueryString(getNextQuery());
    }
  
  
    /**
     * Scans the metadata for the kudu table that this operator is scanning for and
     * returns back the mapping for the kudu column name to the ColumnSchema metadata definition.
     * Note that the Kudu columns names are case sensitive.
     * @return A Map with Kudu column names as keys and value as the Column Definition.
     * @throws Exception
     */
    private Map<String,ColumnSchema> buildColumnSchemaForTable() throws Exception
    {
      if (kuduColNameToSchemaMapping == null) {
        ApexKuduConnection connectionForMetaDataScan = apexKuduConnectionInfo.build();
        KuduTable table = connectionForMetaDataScan.getKuduTable();
        List<ColumnSchema> tableColumns =  table.getSchema().getColumns();
        connectionForMetaDataScan.close();
        Map<String,ColumnSchema> columnSchemaMap = new HashMap<>();
        for (ColumnSchema aColumn: tableColumns) {
          columnSchemaMap.put(aColumn.getName(),aColumn);
        }
        kuduColNameToSchemaMapping = columnSchemaMap;
      }
      return kuduColNameToSchemaMapping;
    }
  
    /**
    * Method responsible for sending a control tuple whenever a new query is about to be processed.
    */
    protected void sendControlTupleForNewQuery()
    {
      C startNewQueryControlTuple = getControlTupleForNewQuery();
      if (startNewQueryControlTuple != null) {
        outputPort.emitControl(startNewQueryControlTuple);
      }
    }
  
    /**
     * Override this method to send a valid Control tuple to the downstream operator instances
     * @return The control tuple that needs to be sent at the beginning of the query
     */
    protected C getControlTupleForNewQuery()
    {
      return null;// default is null; Can be overridden in the child classes if custom representation is needed
    }
  
    /**
     * Fetches values form the buffer. This also checks for the limit set as the maximum tuples that can be
     * sent downstream in a single window.
     * @return Returns the next Kudu row along with its metadata that needs to be processed for filters
     */
  
    protected KuduRecordWithMeta<T> processNextTuple()
    {
      boolean emitTuple = true;
      KuduRecordWithMeta<T> entryFetchedFromBuffer = null;
      if (maxTuplesPerWindow != -1) {
        // we have an explicit upper window. Hence we need to check and then emit
        if (currentWindowTupleCount >= maxTuplesPerWindow) {
          emitTuple = false;
        }
      }
      if ( emitTuple) {
        try {
          entryFetchedFromBuffer = buffer.poll(100L, TimeUnit.MICROSECONDS);
        } catch (InterruptedException e) {
          LOG.debug("No entry available in the buffer " + e.getMessage(), e);
        }
      }
      return entryFetchedFromBuffer;
    }
  
    /***
     * Used to filter of tuples read from the Kudu scan if it is being replayed in the windows before the
     * reconciling window phase. THis method also invokes the filter check in case the current window is in the
     *  reconciling window phase
     * @param recordWithMeta
     */
    protected void filterTupleBasedOnCurrentState(KuduRecordWithMeta<T> recordWithMeta)
    {
      boolean filter = false;
      if (isCurrentlyInSafeMode) {
        filter = true;
      }
      KuduPartitionScanAssignmentMeta currentRecordMeta = recordWithMeta.getTabletMetadata();
      long currentPositionInScan = recordWithMeta.getPositionInScan();
      if (windowManagerDataForScans.containsKey(currentRecordMeta)) {
        long counterLimitForThisMeta = windowManagerDataForScans.get(currentRecordMeta);
        if ( currentPositionInScan <= counterLimitForThisMeta) {
          // This is the case of a replay and hence do not emit
          filter = true;
        } else {
          windowManagerDataForScans.put(currentRecordMeta,currentPositionInScan);
        }
      } else {
        windowManagerDataForScans.put(currentRecordMeta,currentPositionInScan);
      }
      if ( isCurrentlyInReconcilingMode) {
        // check to see if we can emit based on the buisness logic in a reconciling window processing state
        if ( !isAllowedInReconcilingWindow(recordWithMeta)) {
          filter = true;
        }
      }
      if (!filter) {
        outputPort.emit(recordWithMeta.getThePayload());
        currentWindowTupleCount += 1;
      }
    }
  
    /**
     * Sends either control tuples or data tuples basing on the values being fetched from the buffer.
     */
    @Override
    public void emitTuples()
    {
      if (allScansCompleteForThisQuery) {
        if (processForNextQuery() ) {
          sendControlTupleForNewQuery();
          allScansCompleteForThisQuery = false;
        }
        return;
      }
      KuduRecordWithMeta<T> entryFetchedFromBuffer = processNextTuple();
      if (entryFetchedFromBuffer != null) {
        if ( (!entryFetchedFromBuffer.isEndOfScanMarker()) &&
            (!entryFetchedFromBuffer.isBeginScanMarker())) {
          // we have a valid record
          filterTupleBasedOnCurrentState(entryFetchedFromBuffer);
          return;
        }
        if (entryFetchedFromBuffer.isEndOfScanMarker()) {
          processForEndScanMarker(entryFetchedFromBuffer);
          sendControlTupleForEndQuery();
        }
        if (entryFetchedFromBuffer.isBeginScanMarker()) {
          processForBeginScanMarker(entryFetchedFromBuffer);
        }
      }
    }
  
    /**
     * Sends a control tuple to the downstream operators. This method checks to see if the user wishes to send a custom
     *  control tuple. If yes, a control tuple that the user chooses to send is streamed to the downstream operators.
     *  See {@link AbstractKuduInputOperator#getControlTupleForEndQuery()}. If this is null but the user configured a
     *  control tuple to be sent as part of the SQL expression that is given as input, a control tuple will automatically
     *  be generated by this operator containing the message that the user set at the time of the Query definition.
     */
    protected void sendControlTupleForEndQuery()
    {
      if ( currentQueryBeingProcessed == null) {
        return; //
      }
      if (!allScansCompleteForThisQuery) {
        return;
      }
      C endControlTuple = getControlTupleForEndQuery();
      if (endControlTuple != null) {
        outputPort.emitControl(endControlTuple);
      } else {
        if (optionsEnabledForCurrentQuery.containsKey(KuduSQLParseTreeListener.CONTROLTUPLE_MESSAGE)) {
          InputOperatorControlTuple controlTuple = new InputOperatorControlTuple();
          controlTuple.setBeginNewQuery(false);
          controlTuple.setEndCurrentQuery(true);
          controlTuple.setQuery(currentQueryBeingProcessed);
          controlTuple.setControlMessage(
              optionsEnabledForCurrentQuery.get(KuduSQLParseTreeListener.CONTROLTUPLE_MESSAGE));
          outputPort.emitControl(controlTuple);
        }
      }
    }
  
    protected C getControlTupleForEndQuery()
    {
      return null;// default is null; Can be overridden in the child classes if custom representation is needed
    }
  
    protected void processForBeginScanMarker(KuduRecordWithMeta<T> entryFetchedFromBuffer)
    {
      currentQueryCompletionStatus.put(entryFetchedFromBuffer.getTabletMetadata(), false);
    }
  
    /**
     * Used to see if all the segments that are planned for the current query and the current physical instance of the
     * operator are done in terms of streaming tuples to the downstream operators.
     * @param entryFetchedFromBuffer
     */
    protected void processForEndScanMarker(KuduRecordWithMeta<T> entryFetchedFromBuffer)
    {
      Boolean currentStatus = currentQueryCompletionStatus.get(entryFetchedFromBuffer.getTabletMetadata());
      if (currentStatus == null) {
        LOG.error(" End scan marker cannot be precede a Begin Scan marker ");
      }
      currentQueryCompletionStatus.put(entryFetchedFromBuffer.getTabletMetadata(), true);
      if ( plannedSegmentsForCurrentQuery == 0 ) {
        allScansCompleteForThisQuery = true;
        return;
      }
      boolean areAllScansComplete  = true;
      if (currentQueryCompletionStatus.size() != plannedSegmentsForCurrentQuery) {
        return;
      }
      for (KuduPartitionScanAssignmentMeta aMeta: currentQueryCompletionStatus.keySet()) {
        if (!currentQueryCompletionStatus.get(aMeta)) {
          areAllScansComplete = false;
        }
      }
      if (areAllScansComplete) {
        allScansCompleteForThisQuery = true;
      }
    }
  
    /**
     * This method is used to give control the user of the operator to decide how to derive exactly once
     * semantics.  Note that this check is invoked in a reconciling window when the operator is resuming from a crash.
     * This check happens only in this window and in no other window.
     * @param extractedRecord
     * @return true if the tuple needs to be sent to downstream. False if one does not want to stream this tuple
     *  downstream. Note that this check is invoked in a reconciling window when the operator is resuming from a crash.
     */
    protected boolean isAllowedInReconcilingWindow(KuduRecordWithMeta<T> extractedRecord)
    {
      return true;
    }
  
    @Override
    public void beginWindow(long windowId)
    {
      currentWindowTupleCount = 0;
      currentWindowId = windowId;
      if ( currentWindowId != Stateless.WINDOW_ID) {
        if (currentWindowId > reconcilingPhaseWindowId) {
          isCurrentlyInSafeMode = false;
          isCurrentlyInReconcilingMode = false;
        }
        if (currentWindowId == reconcilingPhaseWindowId) {
          isCurrentlyInReconcilingMode = true;
          isCurrentlyInSafeMode = false;
        }
        if (currentWindowId < reconcilingPhaseWindowId) {
          isCurrentlyInReconcilingMode = false;
          isCurrentlyInSafeMode = true;
        }
      }
      LOG.info(" Current processing mode states Safe Mode = " + isCurrentlyInSafeMode + " Reconciling mode = "
          + isCurrentlyInReconcilingMode);
      LOG.info(" Current window ID = " + currentWindowId + " reconciling window ID = " + reconcilingPhaseWindowId);
    }
  
    @Override
    public void endWindow()
    {
      try {
        windowDataManager.save(windowManagerDataForScans,currentWindowId);
      } catch (IOException e) {
        throw new RuntimeException("Could not persist the window Data manager on end window boundary " +
            e.getMessage(),e);
      }
    }
  
    @Override
    public void setup(Context.OperatorContext context)
    {
      if (windowDataManager != null) {
        windowDataManager.setup(context);
      }
      try {
        buildColumnSchemaForTable();
      } catch (Exception e) {
        throw new RuntimeException("Error while trying to build the schema definition for the Kudu table " + tableName,
            e);
      }
      windowManagerDataForScans = new HashMap<>();
      optionsEnabledForCurrentQuery = new HashMap<>();
      initPartitioner();
      initBuffer();
      // Scanner can only be initialized after initializing the partitioner
      initScanner(); // note that this is not streaming any data yet. Only warming up the scanner ready to read data
      initCurrentState();
      isCurrentlyInSafeMode = false;
      isCurrentlyInReconcilingMode = false;
      reconcilingPhaseWindowId = Stateless.WINDOW_ID;
      if ( (context.getValue(Context.OperatorContext.ACTIVATION_WINDOW_ID) != Stateless.WINDOW_ID) &&
          context.getValue(Context.OperatorContext.ACTIVATION_WINDOW_ID) <
          windowDataManager.getLargestCompletedWindow()) {
        isCurrentlyInSafeMode = true;
        reconcilingPhaseWindowId = windowDataManager.getLargestCompletedWindow() + 1;
        LOG.info("Set reconciling window ID as " + reconcilingPhaseWindowId);
        isCurrentlyInReconcilingMode = false;
      }
    }
  
    @Override
    public void teardown()
    {
      scanner.close();
    }
  
    /**
     * Used to read an existing state from a window data manager and initialize the starting state of this operator. Note
     * that there might be a query that was partially processed before the checkpointing process or the crash was
     *  triggered.
     */
    private void initCurrentState()
    {
      Map<KuduPartitionScanAssignmentMeta,Long> savedState = null;
      if ( (windowManagerDataForScans.size() == 0) && (currentQueryBeingProcessed == null) ) {
        // This is the case of an application restart possibly and hence want to get the exact state
        try {
          savedState = (Map<KuduPartitionScanAssignmentMeta,Long>)windowDataManager.retrieve(
            windowDataManager.getLargestCompletedWindow());
        } catch (IOException e) {
          throw new RuntimeException("Error while retrieving the window manager data at the initialization phase", e);
        } catch ( NullPointerException ex) {
          LOG.error("Error while getting the window manager data ", ex);
        }
      }
      if ( ( savedState != null) && (savedState.size() > 0 ) ) {
        KuduPartitionScanAssignmentMeta aMeta = savedState.keySet().iterator().next(); // we have one atleast
        currentQueryBeingProcessed = aMeta.getCurrentQuery();
        allScansCompleteForThisQuery = false;
        windowManagerDataForScans.putAll(savedState);
        processForQueryString(currentQueryBeingProcessed);
      }
    }
  
    @Override
    public Collection<Partition<AbstractKuduInputOperator>> definePartitions(Collection collection,
        PartitioningContext context)
    {
      initPartitioner();
      return partitioner.definePartitions(collection,context);
    }
  
    @Override
    public void partitioned(Map partitions)
    {
      initPartitioner();
      partitioner.partitioned(partitions);
    }
  
    @Override
    public Response processStats(BatchedOperatorStats stats)
    {
      Response response = new Response();
      if ( ( partitionPieAssignment == null) || (partitionPieAssignment.size() == 0) ) {
        response.repartitionRequired = true;
      } else {
        response.repartitionRequired = false;
      }
      return response;
    }
  
    private void initBuffer()
    {
      buffer = new DisruptorBlockingQueue<KuduRecordWithMeta<T>>(bufferCapacity,cpuSpinPolicyForWaitingInBuffer);
    }
  
    private void initPartitioner()
    {
      if (partitioner != null) {
        return;
      }
      checkNotNull(apexKuduConnectionInfo,"Apex Kudu connection cannot be null while setting partitioner");
      switch (partitionScanStrategy) {
        case MANY_TABLETS_PER_OPERATOR:
          partitioner = new KuduOneToManyPartitioner(this);
          break;
        case ONE_TABLET_PER_OPERATOR:
        default:
          partitioner = new KuduOneToOnePartitioner(this);
          break;
  
      }
    }
  
    private void initScanner()
    {
      switch (scanOrderStrategy) {
        case CONSISTENT_ORDER_SCANNER:
          setFaultTolerantScanner(true); // Consistent order scanners require them to be fault tolerant
          scanner = new KuduPartitionConsistentOrderScanner<>(this);
          break;
        case RANDOM_ORDER_SCANNER:
        default:
          scanner = new KuduPartitionRandomOrderScanner(this);
          break;
      }
    }
  
    /***
     *
     * @param parsedQuery The parsed query string
     * @return Null if the SQL expression cannot be mapped properly to the POJO fields in which case the SQL string is
     *  sent to the Error port
     */
    public Map<String,Object> extractSettersForResultObject(SQLToKuduPredicatesTranslator parsedQuery)
    {
      Map<String,String> aliasesUsedForThisQuery = parsedQuery.getKuduSQLParseTreeListener().getAliases();
      Map<String,Object> setterMap = new HashMap<>();
      Field[] fieldsOfThisPojo = clazzForResultObject.getDeclaredFields();
      Set<String> allPojoFieldNamesUsed = new HashSet<>(aliasesUsedForThisQuery.values());
      for ( Field aField : fieldsOfThisPojo) {
        if (!allPojoFieldNamesUsed.contains(aField.getName())) {
          LOG.error("Invalid mapping fo Kudu table column name to the POJO field name " + aField.getName());
          return null; // SQL expression will be sent to the error port
        }
      }
      for (ColumnSchema aKuduTableColumn: kuduColNameToSchemaMapping.values()) {
        String kuduColumnName = aKuduTableColumn.getName();
        switch ( aKuduTableColumn.getType().getDataType().getNumber()) {
          case Common.DataType.BINARY_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetter(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName),ByteBuffer.class));
            break;
          case Common.DataType.BOOL_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetterBoolean(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName)));
            break;
          case Common.DataType.DOUBLE_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetterDouble(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName)));
            break;
          case Common.DataType.FLOAT_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetterFloat(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName)));
            break;
          case Common.DataType.INT8_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetterByte(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName)));
            break;
          case Common.DataType.INT16_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetterShort(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName)));
            break;
          case Common.DataType.INT32_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetterInt(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName)));
            break;
          case Common.DataType.UNIXTIME_MICROS_VALUE:
          case Common.DataType.INT64_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetterLong(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName)));
            break;
          case Common.DataType.STRING_VALUE:
            setterMap.put(kuduColumnName,PojoUtils.createSetter(clazzForResultObject,
                aliasesUsedForThisQuery.get(kuduColumnName),String.class));
            break;
          case Common.DataType.UINT8_VALUE:
            LOG.error("Unsigned int 8 not supported yet");
            throw new RuntimeException("uint8 not supported in Kudu schema yet");
          case Common.DataType.UINT16_VALUE:
            LOG.error("Unsigned int 16 not supported yet");
            throw new RuntimeException("uint16 not supported in Kudu schema yet");
          case Common.DataType.UINT32_VALUE:
            LOG.error("Unsigned int 32 not supported yet");
            throw new RuntimeException("uint32 not supported in Kudu schema yet");
          case Common.DataType.UINT64_VALUE:
            LOG.error("Unsigned int 64 not supported yet");
            throw new RuntimeException("uint64 not supported in Kudu schema yet");
          case Common.DataType.UNKNOWN_DATA_VALUE:
            LOG.error("unknown data type ( complex types ? )  not supported yet");
            throw new RuntimeException("Unknown data type  ( complex types ? ) not supported in Kudu schema yet");
          default:
            LOG.error("unknown type/default  ( complex types ? )  not supported yet");
            throw new RuntimeException("Unknown type/default  ( complex types ? ) not supported in Kudu schema yet");
        }
      }
      return setterMap;
    }
  
    @Override
    public void activate(Context.OperatorContext context)
    {
    }
  
    @Override
    public void deactivate()
    {
  
    }
  
    @Override
    public void beforeCheckpoint(long windowId)
    {
  
    }
  
    @Override
    public void checkpointed(long windowId)
    {
  
    }
  
    @Override
    public void committed(long windowId)
    {
  
    }
  
    public ApexKuduConnection.ApexKuduConnectionBuilder getApexKuduConnectionInfo()
    {
      return apexKuduConnectionInfo;
    }
  
    public void setApexKuduConnectionInfo(ApexKuduConnection.ApexKuduConnectionBuilder apexKuduConnection)
    {
      apexKuduConnectionInfo = apexKuduConnection;
    }
  
    public List<KuduPartitionScanAssignmentMeta> getPartitionPieAssignment()
    {
      return partitionPieAssignment;
    }
  
    public void setPartitionPieAssignment(List<KuduPartitionScanAssignmentMeta> partitionPieAssignment)
    {
      this.partitionPieAssignment = partitionPieAssignment;
    }
  
    public KuduPartitionScanStrategy getPartitionScanStrategy()
    {
      return partitionScanStrategy;
    }
  
    public void setPartitionScanStrategy(KuduPartitionScanStrategy partitionScanStrategy)
    {
      this.partitionScanStrategy = partitionScanStrategy;
    }
  
    public String getCurrentQueryBeingProcessed()
    {
      return currentQueryBeingProcessed;
    }
  
    public void setCurrentQueryBeingProcessed(String currentQueryBeingProcessed)
    {
      this.currentQueryBeingProcessed = currentQueryBeingProcessed;
    }
  
    public int getNumberOfPartitions()
    {
      return numberOfPartitions;
    }
  
    public void setNumberOfPartitions(int numberOfPartitions)
    {
      this.numberOfPartitions = numberOfPartitions;
    }
  
    public KuduScanOrderStrategy getScanOrderStrategy()
    {
      return scanOrderStrategy;
    }
  
    public void setScanOrderStrategy(KuduScanOrderStrategy scanOrderStrategy)
    {
      this.scanOrderStrategy = scanOrderStrategy;
    }
  
    public AbstractKuduInputPartitioner getPartitioner()
    {
      return partitioner;
    }
  
    public void setPartitioner(AbstractKuduInputPartitioner partitioner)
    {
      this.partitioner = partitioner;
    }
  
    public AbstractKuduPartitionScanner<T,C> getScanner()
    {
      return scanner;
    }
  
    public void setScanner(AbstractKuduPartitionScanner<T,C> scanner)
    {
      this.scanner = scanner;
    }
  
    public Class<T> getClazzForResultObject()
    {
      return clazzForResultObject;
    }
  
    public void setClazzForResultObject(Class<T> clazzForResultObject)
    {
      this.clazzForResultObject = clazzForResultObject;
    }
  
    public String getTableName()
    {
      return tableName;
    }
  
    public void setTableName(String tableName)
    {
      this.tableName = tableName;
    }
  
    public SpinPolicy getCpuSpinPolicyForWaitingInBuffer()
    {
      return cpuSpinPolicyForWaitingInBuffer;
    }
  
    public void setCpuSpinPolicyForWaitingInBuffer(SpinPolicy cpuSpinPolicyForWaitingInBuffer)
    {
      this.cpuSpinPolicyForWaitingInBuffer = cpuSpinPolicyForWaitingInBuffer;
    }
  
    public int getBufferCapacity()
    {
      return bufferCapacity;
    }
  
    public void setBufferCapacity(int bufferCapacity)
    {
      this.bufferCapacity = bufferCapacity;
    }
  
    public Map<String, String> getKuduColNameToPOJOFieldNameMap()
    {
      return kuduColNameToPOJOFieldNameMap;
    }
  
    public void setKuduColNameToPOJOFieldNameMap(Map<String, String> kuduColNameToPOJOFieldNameMap)
    {
      this.kuduColNameToPOJOFieldNameMap = kuduColNameToPOJOFieldNameMap;
    }
  
    public Map<String, ColumnSchema> getKuduColNameToSchemaMapping()
    {
      return kuduColNameToSchemaMapping;
    }
  
    public void setKuduColNameToSchemaMapping(Map<String, ColumnSchema> kuduColNameToSchemaMapping)
    {
      this.kuduColNameToSchemaMapping = kuduColNameToSchemaMapping;
    }
  
    public int getMaxTuplesPerWindow()
    {
      return maxTuplesPerWindow;
    }
  
    public void setMaxTuplesPerWindow(int maxTuplesPerWindow)
    {
      this.maxTuplesPerWindow = maxTuplesPerWindow;
    }
  
    /**
     * Note that it is recommended that the buffer capacity be in the powers of two
     * @param buffer
     */
    public void setBuffer(DisruptorBlockingQueue<KuduRecordWithMeta<T>> buffer)
    {
      this.buffer = buffer;
    }
  
    public DisruptorBlockingQueue<KuduRecordWithMeta<T>> getBuffer()
    {
      return buffer;
    }
  
    public WindowDataManager getWindowDataManager()
    {
      return windowDataManager;
    }
  
    public void setWindowDataManager(WindowDataManager windowDataManager)
    {
      this.windowDataManager = windowDataManager;
    }
  
    public boolean isFaultTolerantScanner()
    {
      return isFaultTolerantScanner;
    }
  
    public void setFaultTolerantScanner(boolean faultTolerantScanner)
    {
      isFaultTolerantScanner = faultTolerantScanner;
    }
  
    public Map<String, String> getOptionsEnabledForCurrentQuery()
    {
      return optionsEnabledForCurrentQuery;
    }
  
    public void setOptionsEnabledForCurrentQuery(Map<String, String> optionsEnabledForCurrentQuery)
    {
      this.optionsEnabledForCurrentQuery = optionsEnabledForCurrentQuery;
    }
  
    public boolean isPartitioned()
    {
      return isPartitioned;
    }
  
    public void setPartitioned(boolean partitioned)
    {
      isPartitioned = partitioned;
    }
  }
  