gemfire-core/src/test/java/com/gemstone/gemfire/internal/datasource/ConnectionPoolingJUnitTest.java

/*=========================================================================
 * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
 * This product is protected by U.S. and international copyright
 * and intellectual property laws. Pivotal products are covered by
 * one or more patents listed at http://www.pivotal.io/patents.
 *=========================================================================
 */
package com.gemstone.gemfire.internal.datasource;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.fail;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import javax.naming.Context;
import javax.naming.NamingException;
import javax.sql.DataSource;
import javax.transaction.UserTransaction;

import org.apache.logging.log4j.Logger;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import com.gemstone.gemfire.cache.Cache;
import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.gemfire.distributed.DistributedSystem;
import com.gemstone.gemfire.distributed.internal.DistributionConfig;
import com.gemstone.gemfire.internal.logging.LogService;
import com.gemstone.gemfire.util.test.TestUtil;
import com.gemstone.gemfire.test.junit.categories.IntegrationTest;

/**
 * @author Mitul
 * 
 *  
 */
@Category(IntegrationTest.class)
public class ConnectionPoolingJUnitTest {
  private static final Logger logger = LogService.getLogger();

  private Thread ThreadA;
  protected Thread ThreadB;
//  private Thread ThreadC;
  protected static int maxPoolSize = 7;
  protected static DataSource ds = null;
  private static Properties props = null;
  private static DistributedSystem ds1 = null;
  protected static Cache cache = null;
  protected boolean encounteredException = false;
  
  @Before
  public void setUp() throws Exception {
    encounteredException = false;
    props = new Properties();
    props.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
    String path = TestUtil.getResourcePath(ConnectionPoolingJUnitTest.class, "/jta/cachejta.xml");
    props.setProperty("cache-xml-file", path);
    ds1 = DistributedSystem.connect(props);
    cache = CacheFactory.create(ds1);
  }

  @After
  public void tearDown() throws Exception {
    ds1.disconnect();
  }

  @Test
  public void testGetSimpleDataSource() throws Exception {
    Context ctx = cache.getJNDIContext();
    GemFireBasicDataSource ds = (GemFireBasicDataSource) ctx.lookup("java:/SimpleDataSource");
    Connection conn = ds.getConnection();
    if (conn == null) {
      fail("DataSourceFactoryTest-testGetSimpleDataSource() Error in creating the GemFireBasicDataSource");
    }
  }

  /*
   * public static void main(String str[]) { ConnectionPoolingJUnitTest test = new
   * ConnectionPoolingJUnitTest(); test.setup(); try {
   * test.testGetSimpleDataSource(); } catch (Exception e) { // TODO
   * Auto-generated catch block e.printStackTrace(); fail(); }
   * test.testConnectionPoolFunctions(); test.teardown();
   *  
   */
  @Test
  public void testConnectionPoolFunctions() throws Exception {
    Context ctx = cache.getJNDIContext();
    ds = (GemFireConnPooledDataSource) ctx.lookup("java:/PooledDataSource");
    PoolClient_1 clientA = new PoolClient_1();
    ThreadA = new Thread(clientA, "ThreadA");
    PoolClient_2 clientB = new PoolClient_2();
    ThreadB = new Thread(clientB, "ThreadB");
    // ThreadA.setDaemon(true);
    //ThreadB.setDaemon(true);
    ThreadA.start();
  }

  @Ignore("Disabled for bug #52242")
  @Test
  public void testXAPoolLeakage() throws Exception {
    final String tableName = "testXAPoolLeakage";
    //initialize cache and get user transaction
    try {
      int numThreads = 10;
      final int LOOP_COUNT = 10;

      //logger.debug ("Table name: " + tableName);
      Context ctx = cache.getJNDIContext();
      DataSource ds = (DataSource)ctx.lookup("java:/SimpleDataSource");

      //String sql = "create table " + tableName + " (id number primary key,
      // name varchar2(50))";
      //String sql = "create table " + tableName + " (id integer primary key,
      // name varchar(50))";
      String sql = "create table "
          + tableName
          + " (id varchar(50) NOT NULL, name varchar(50), CONSTRAINT "+tableName+"_key PRIMARY KEY(id))";
      logger.debug(sql);
      Connection conn = ds.getConnection();
      Statement sm = conn.createStatement();
      sm.execute(sql);
      sm.close();
      conn.close();
      Thread th[] = new Thread[numThreads];
      for (int i = 0; i < numThreads; ++i) {
        final int threadID = i;
        th[i] = new Thread(new Runnable() {
          private int key = threadID;

          public void run()
          {
            try {
              Context ctx = cache.getJNDIContext();
              //          Operation with first XA Resource
              DataSource da1 = (DataSource)ctx
                  .lookup("java:/XAMultiThreadedDataSource");
               int val=0;
              for (int j = 0; j < LOOP_COUNT; ++j) {
                UserTransaction ta = null;
                try {
                  ta = (UserTransaction)ctx.lookup("java:/UserTransaction");

                }
                catch (NamingException e) {
                  encounteredException = true;
                  break;
                }

                try {
                  //Begin the user transaction
                  ta.begin();
                  for (int i = 1; i <= 50; i++) {
                    Connection conn = da1.getConnection();
                    Statement sm = conn.createStatement();

                    String sql = "insert into " + tableName + " values (" + "'"
                        + key + "X" + ++val + "','name" + i + "')";                    
                    sm.execute(sql);
                 
                    sm.close();
                    conn.close();
                  }
                  if(j%2 == 0) {
                    ta.commit();
                    logger.debug("Committed successfully for thread with id ="+key);
                  }else{
                    ta.rollback();
                    logger.debug("Rolled back successfully for thread with id ="+key);
                  }
                }
                catch (Exception e) {
                  e.printStackTrace();
                  encounteredException = true;
                  break;
                }

              }
            }
            catch (Exception e) {
              e.printStackTrace();
              encounteredException = true;
            }
          }
        });
      }

      for (int i = 0; i < th.length; ++i) {
        th[i].start();
      }

      for (int i = 0; i < th.length; ++i) {
        Thread t = th[i];
        long ms = 90 * 1000;
        t.join(ms);
        if (t.isAlive()) {
          for(int j =0; j < th.length; j++) {
            th[j].interrupt();
          }
          fail("Thread did not terminate after " + ms + " ms: " + t);
        }
      }
      assertFalse(encounteredException);

    } finally {
      logger.debug("Destroying table: " + tableName);

      Context ctx = cache.getJNDIContext();
      DataSource ds = (DataSource)ctx.lookup("java:/SimpleDataSource");
      Connection conn = ds.getConnection();
      logger.debug (" trying to drop table: " + tableName);
      String sql = "drop table " + tableName;
      Statement sm = conn.createStatement();
      sm.execute(sql);
      conn.close();
      logger.debug (" Dropped table: " + tableName);
    }
  }

  private class PoolClient_1 implements Runnable {
    public void run() {
      String threadName = Thread.currentThread().getName();
      logger.debug(" Inside Run method of " + threadName);
      int numConn = 0;
      Object[] connections = new Object[maxPoolSize];
//      Statement stmt = null;
      while (numConn < maxPoolSize) {
        try {
          logger.debug(" Getting a connection from " + threadName);
          //	ds.getConnection();
          connections[numConn] = ds.getConnection();
          //   Thread.sleep(500);
          numConn++;
          //if (numConn == 5)
          //ThreadB.start();
          logger.debug(" Got connection " + numConn + "from " + threadName);
        }
        catch (Exception ex) {
          ex.printStackTrace();
        }
      }
      if (numConn != maxPoolSize) fail("#### Error in filling the the connection pool from " + threadName);
      ThreadB.start();
      logger.debug(" AFTER starting THREADB");
      int numC = 0;
      int display = 0;
//      long birthTime = 0;
//      long newTime = 0;
//      long duration = 0;
      Connection conn;
      try {
        Thread.sleep(500);
      }
      catch (InterruptedException e) {
        e.printStackTrace();
        fail("interrupted");
      }
      for (numC = 0; numC < numConn; numC++) {
        try {
          display = numC + 1;
          logger.debug(" Returning connection " + display + "from " + threadName);
          conn = (Connection) connections[numC];
          logger.debug(" ************************************" + conn);
          logger.debug(" !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! The connection is of type "
              + conn.getClass());
          //   goahead = false;
          // conn.close();
          logger.debug(" Returned connection " + display + "from " + threadName);
        }
        catch (Exception ex) {
          fail("Exception occured in trying to returnPooledConnectiontoPool due to " + ex);
          ex.printStackTrace();
        }
      }
      if (numC != maxPoolSize) fail("#### Error in returning all the connections to the  pool from " + threadName);
      logger.debug(" ****************Returned all connections " + threadName + "***********");
    }
  }

  private static class PoolClient_2 implements Runnable {

    List poolConnlist = new ArrayList();

    public void run() {
      try {
        String threadName = Thread.currentThread().getName();
        logger.debug(" Inside Run method of " + threadName);
        int numConn2 = 0;
//        int display = 0;
        Object[] connections = new Object[maxPoolSize];
        while (numConn2 < maxPoolSize) {
          try {
            logger.debug(" _______________________________________________________________ " + numConn2);
            numConn2++;
            logger.debug(" ********** Before getting " + numConn2 + "from" + threadName);
            connections[numConn2 - 1] = ds.getConnection();
            logger.debug(" ********** Got connection " + numConn2 + "from" + threadName);
          }
          catch (Exception ex) {
            ex.printStackTrace();
          }
        }
        numConn2 = 0;
        while (numConn2 < maxPoolSize) {
          try {
            ((Connection) connections[numConn2]).close();
            numConn2++;
          }
          catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
          }
        }
        if (numConn2 != maxPoolSize) fail("#### Error in getting all connections from the " + threadName);
        logger.debug(" ****************GOT ALL connections " + threadName + "***********");
      }
      catch (Exception e1) {
        e1.printStackTrace();
        fail();
      }
    }
  }
}