contrib/src/main/java/org/apache/apex/malhar/contrib/accumulo/AccumuloWindowStore.java

/**
 * 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.contrib.accumulo;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.apex.malhar.lib.db.TransactionableStore;
import org.apache.hadoop.io.Text;

import com.datatorrent.netlet.util.DTThrowable;

/**
 * Provides transactional support by implementing TransactionableStore abstract methods.
 *
 * <p>
 * Not intended for true transactional
 * properties. It does not guarantee exactly once property.It only skips tuple
 * processed in previous windows
 *
 * @displayName Accumulo Window Store
 * @category Output
 * @tags accumulo, key value
 * @since 1.0.4
 */
public class AccumuloWindowStore extends AccumuloStore implements TransactionableStore
{
  private static final transient Logger logger = LoggerFactory.getLogger(AccumuloWindowStore.class);
  private static final String DEFAULT_ROW_NAME = "AccumuloOperator_row";
  private static final String DEFAULT_COLUMN_FAMILY_NAME = "AccumuloOutputOperator_cf";
  private static final String DEFAULT_LAST_WINDOW_PREFIX_COLUMN_NAME = "last_window";
  private transient String rowName;
  private transient String columnFamilyName;

  private transient byte[] rowBytes;
  private transient byte[] columnFamilyBytes;

  private transient String lastWindowColumnName;
  private transient byte[] lastWindowColumnBytes;

  public AccumuloWindowStore()
  {
    rowName = DEFAULT_ROW_NAME;
    columnFamilyName = DEFAULT_COLUMN_FAMILY_NAME;
    lastWindowColumnName = DEFAULT_LAST_WINDOW_PREFIX_COLUMN_NAME;
    constructKeys();
  }

  /**
   * the values are stored as byte arrays.This method converts string to byte
   * arrays. uses util class in hbase library to do so.
   */
  private void constructKeys()
  {
    rowBytes = rowName.getBytes();
    columnFamilyBytes = columnFamilyName.getBytes();
  }

  public String getRowName()
  {
    return rowName;
  }

  public void setRowName(String rowName)
  {
    this.rowName = rowName;
    constructKeys();
  }

  public String getColumnFamilyName()
  {
    return columnFamilyName;
  }

  public void setColumnFamilyName(String columnFamilyName)
  {
    this.columnFamilyName = columnFamilyName;
    constructKeys();
  }

  @Override
  public void beginTransaction()
  {
    // accumulo does not support transactions
  }

  @Override
  public void commitTransaction()
  {
    // accumulo does not support transactions

  }

  @Override
  public void rollbackTransaction()
  {
    // accumulo does not support transactions

  }

  @Override
  public boolean isInTransaction()
  {
    // accumulo does not support transactions
    return false;
  }

  @Override
  public long getCommittedWindowId(String appId, int operatorId)
  {
    byte[] value = null;
    Authorizations auths = new Authorizations();
    Scanner scan = null;
    String columnKey = appId + "_" + operatorId + "_" + lastWindowColumnName;
    lastWindowColumnBytes = columnKey.getBytes();
    try {
      scan = connector.createScanner(tableName, auths);
    } catch (TableNotFoundException e) {
      logger.error("error getting committed window id", e);
      DTThrowable.rethrow(e);
    }
    scan.setRange(new Range(new Text(rowBytes)));
    scan.fetchColumn(new Text(columnFamilyBytes), new Text(lastWindowColumnBytes));
    for (Entry<Key, Value> entry : scan) {
      value = entry.getValue().get();
    }
    if (value != null) {
      long longval = toLong(value);
      return longval;
    }
    return -1;
  }

  @Override
  public void storeCommittedWindowId(String appId, int operatorId,long windowId)
  {
    byte[] WindowIdBytes = toBytes(windowId);
    String columnKey = appId + "_" + operatorId + "_" + lastWindowColumnName;
    lastWindowColumnBytes = columnKey.getBytes();
    Mutation mutation = new Mutation(rowBytes);
    mutation.put(columnFamilyBytes, lastWindowColumnBytes, WindowIdBytes);
    try {
      batchwriter.addMutation(mutation);
      batchwriter.flush();
    } catch (MutationsRejectedException e) {
      logger.error("error getting committed window id", e);
      DTThrowable.rethrow(e);
    }
  }

  @Override
  public void removeCommittedWindowId(String appId, int operatorId)
  {
    // accumulo does not support transactions
  }

  public static byte[] toBytes(long l)
  {
    ByteArrayOutputStream baos = new ByteArrayOutputStream(Long.SIZE / 8);
    DataOutputStream dos = new DataOutputStream(baos);
    byte[] result = null;
    try {
      dos.writeLong(l);
      result = baos.toByteArray();
      dos.close();
    } catch (IOException e) {
      logger.error("error converting to byte array");
      DTThrowable.rethrow(e);
    }
    return result;
  }

  public static long toLong(byte[] b)
  {
    ByteArrayInputStream baos = new ByteArrayInputStream(b);
    DataInputStream dos = new DataInputStream(baos);
    long result = 0;
    try {
      result = dos.readLong();
      dos.close();
    } catch (IOException e) {
      logger.error("error converting to long");
      DTThrowable.rethrow(e);
    }
    return result;
  }

}