Mailing-List: contact dev-help@apex.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@apex.apache.org
From: chinmaykolhatkar <git@git.apache.org>
To: dev@apex.incubator.apache.org
Reply-To: dev@apex.incubator.apache.org
References: <git-pr-292-apex-malhar@git.apache.org>
In-Reply-To: <git-pr-292-apex-malhar@git.apache.org>
Subject: [GitHub] apex-malhar pull request #292: APEXMALHAR-1957: Updated HBase Input Operator
Content-Type: text/plain
Message-Id: <20160613232042.D4804DFB74@git1-us-west.apache.org>
Date: Mon, 13 Jun 2016 23:20:42 +0000 (UTC)
archived-at: Mon, 13 Jun 2016 23:20:47 -0000

Github user chinmaykolhatkar commented on a diff in the pull request:

    https://github.com/apache/apex-malhar/pull/292#discussion_r66885954
  
    --- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
    @@ -40,27 +51,107 @@
      * @tags hbase, scan, input operator
      * @since 0.3.2
      */
    -public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T>
    +public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T> implements Operator.ActivationListener<Context>
     {
    +  public static final int DEF_HINT_SCAN_LOOKAHEAD = 2;
    +  public static final int DEF_QUEUE_SIZE = 1000;
    +  public static final int DEF_SLEEP_MILLIS = 10;
    +
    +  private String startRow;
    +  private String endRow;
    +  private String lastReadRow;
    +  private int hintScanLookahead = DEF_HINT_SCAN_LOOKAHEAD;
    +  private int queueSize = DEF_QUEUE_SIZE;
    +  private int sleepMillis = DEF_SLEEP_MILLIS;
    +  private Queue<Result> resultQueue;
    +  private boolean threadFailed = false;
    +
    +  @AutoMetric
    +  protected long tuplesRead;
    +
    +  // Transients
    +  protected transient Scan scan;
    +  protected transient ResultScanner scanner;
    +  protected transient Thread readThread;
    +
    +  @Override
    +  public void setup(OperatorContext context)
    +  {
    +    super.setup(context);
    +    resultQueue = Queues.newLinkedBlockingQueue(queueSize);
    +  }
    +
    +  @Override
    +  public void activate(Context context)
    +  {
    +    startReadThread();
    +  }
    +
    +  protected void startReadThread()
    +  {
    +    try {
    +      scan = operationScan();
    +      scanner = getStore().getTable().getScanner(scan);
    +    } catch (IOException e) {
    +      throw new RuntimeException(e);
    +    }
    +    readThread = new Thread(new Runnable() {
    +      @Override
    +      public void run()
    +      {
    +        try {
    +          Result result;
    +          while ((result = scanner.next()) != null) {
    +            while (!resultQueue.offer(result)) {
    +              Thread.sleep(sleepMillis);
    +            }
    +          }
    +        } catch (Exception e) {
    +          logger.debug("Exception in fetching results {}", e.getMessage());
    +          threadFailed = true;
    +          throw new RuntimeException(e);
    +        } finally {
    +          scanner.close();
    +        }
    +      }
    +    });
    +    readThread.start();
    +  }
    +
    +  @Override
    +  public void beginWindow(long windowId)
    +  {
    +    super.beginWindow(windowId);
    +    tuplesRead = 0;
    +  }
     
       @Override
       public void emitTuples()
       {
    +    if (!readThread.isAlive() && threadFailed) {
    +      throw new RuntimeException("Exception in scan thread");
    +    }
         try {
    -      HTable table = getTable();
    -      Scan scan = operationScan();
    -      ResultScanner scanner = table.getScanner(scan);
    -      for (Result result : scanner) {
    -        //KeyValue[] kvs = result.raw();
    -        //T t = getTuple(kvs);
    -        T t = getTuple(result);
    -        outputPort.emit(t);
    +      Result result = resultQueue.poll();
    +      if (result == null) {
    +        return;
    +      }
    +      T tuple = getTuple(result);
    +      if (tuple != null) {
    +        outputPort.emit(tuple);
    +        tuplesRead++;
           }
         } catch (Exception e) {
    -      e.printStackTrace();
    +      throw new RuntimeException(e);
         }
       }
     
    +  @Override
    +  public void deactivate()
    +  {
    +    readThread.interrupt();
    --- End diff --
    
    Please don't interrupt I would suggest. Have a variable in thread which is set to true for running and false for existing. Then wait for the operation to finish.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---