Mailing-List: contact dev-help@apex.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@apex.apache.org
Date: Mon, 8 Aug 2016 08:53:20 +0000 (UTC)
From: "ASF GitHub Bot (JIRA)" <jira@apache.org>
To: dev@apex.incubator.apache.org
Message-ID: <JIRA.12993489.1469784242000.247230.1470646400533@Atlassian.JIRA>
In-Reply-To: <JIRA.12993489.1469784242000@Atlassian.JIRA>
References: <JIRA.12993489.1469784242000@Atlassian.JIRA> <JIRA.12993489.1469784242142@arcas>
Subject: [jira] [Commented] (APEXMALHAR-2172) Update JDBC poll input
 operator to fix issues
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Mon, 08 Aug 2016 08:53:25 -0000


    [ https://issues.apache.org/jira/browse/APEXMALHAR-2172?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15411521#comment-15411521 ] 

ASF GitHub Bot commented on APEXMALHAR-2172:
--------------------------------------------

Github user DT-Priyanka commented on a diff in the pull request:

    https://github.com/apache/apex-malhar/pull/358#discussion_r73841414
  
    --- Diff: library/src/main/java/com/datatorrent/lib/db/jdbc/AbstractJdbcPollInputOperator.java ---
    @@ -438,119 +261,110 @@ public void endWindow()
         currentWindowRecoveryState = new MutablePair<>();
       }
     
    -  public int getPartitionCount()
    -  {
    -    return partitionCount;
    -  }
    -
    -  public void setPartitionCount(int partitionCount)
    +  @Override
    +  public void deactivate()
       {
    -    this.partitionCount = partitionCount;
    +    scanService.shutdownNow();
    +    store.disconnect();
       }
     
    -  @Override
    -  public void activate(Context cntxt)
    +  protected void pollRecords()
       {
    -    if (context.getValue(OperatorContext.ACTIVATION_WINDOW_ID) != Stateless.WINDOW_ID
    -        && context.getValue(OperatorContext.ACTIVATION_WINDOW_ID) < windowManager.getLargestRecoveryWindow()) {
    -      // If it is a replay state, don't start any threads here
    +    if (isPolled) {
           return;
         }
    -  }
    -
    -  @Override
    -  public void deactivate()
    -  {
         try {
    -      if (dbPoller != null && dbPoller.isAlive()) {
    -        dbPoller.interrupt();
    -        dbPoller.join();
    +      ps.setFetchSize(getFetchSize());
    +      ResultSet result = ps.executeQuery();
    +      if (result.next()) {
    +        do {
    +          emitQueue.add(getTuple(result));
    +        } while (result.next());
           }
    -    } catch (InterruptedException ex) {
    -      // log and ignore, ending execution anyway
    -      LOG.error("exception in poller thread: ", ex);
    +      isPolled = true;
    +    } catch (SQLException ex) {
    +      throw new RuntimeException(String.format("Error while running query"), ex);
    +    } finally {
    +      store.disconnect();
         }
    -  }
     
    -  @Override
    -  public void handleIdleTime()
    -  {
    -    if (execute) {
    -      try {
    -        Thread.sleep(spinMillis);
    -      } catch (InterruptedException ie) {
    -        throw new RuntimeException(ie);
    -      }
    -    } else {
    -      LOG.error("Exception: ", cause);
    -      DTThrowable.rethrow(cause.get());
    -    }
       }
     
    +  public abstract T getTuple(ResultSet result);
    +
       protected void replay(long windowId) throws SQLException
       {
    -    isReplayed = true;
     
    -    MutablePair<String, String> recoveredData = new MutablePair<String, String>();
         try {
    -      recoveredData = (MutablePair<String, String>)windowManager.load(operatorId, windowId);
    +      MutablePair<Integer, Integer> recoveredData = (MutablePair<Integer, Integer>)windowManager.load(operatorId,
    +          windowId);
     
    -      if (recoveredData != null) {
    -        //skip the window and return if there was no incoming data in the window
    -        if (recoveredData.left == null || recoveredData.right == null) {
    -          return;
    -        }
    -
    -        if (recoveredData.right.equals(rangeQueryPair.getValue()) || recoveredData.right.equals(previousUpperBound)) {
    -          LOG.info("Matched so returning");
    -          return;
    -        }
    +      if (recoveredData != null && shouldReplayWindow(recoveredData)) {
    +        LOG.debug("[Recovering Window ID - {} for record range: {}, {}]", windowId, recoveredData.left,
    +            recoveredData.right);
     
    -        JdbcPollInputOperator jdbcPoller = new JdbcPollInputOperator();
    -        jdbcPoller.setStore(store);
    -        jdbcPoller.setKey(getKey());
    -        jdbcPoller.setPartitionCount(getPartitionCount());
    -        jdbcPoller.setPollInterval(getPollInterval());
    -        jdbcPoller.setTableName(getTableName());
    -        jdbcPoller.setBatchSize(getBatchSize());
    -        isPollable = false;
    -
    -        LOG.debug("[Window ID -" + windowId + "," + recoveredData.left + "," + recoveredData.right + "]");
    +        ps = store.getConnection().prepareStatement(
    +            buildRangeQuery(rangeQueryPair.getKey(), (rangeQueryPair.getValue() - rangeQueryPair.getKey())),
    --- End diff --
    
    right, would update.


> Update JDBC poll input operator to fix issues
> ---------------------------------------------
>
>                 Key: APEXMALHAR-2172
>                 URL: https://issues.apache.org/jira/browse/APEXMALHAR-2172
>             Project: Apache Apex Malhar
>          Issue Type: Improvement
>            Reporter: Priyanka Gugale
>            Assignee: Priyanka Gugale
>
> Update JDBCPollInputOperator to:
> 1. Fix small bugs
> 2. Use jooq query dsl library to construct sql queries
> 3. Make code more readable
> 4. Use row counts rather than key column values to partition reads


--
This message was sent by Atlassian JIRA
(v6.3.4#6332)