AuditDBHelper { public Set<Tuple> retrieve(Date toDate, Date fromDate, Filter filter, GroupBy groupBy) { LOG.debug("Retrieving from db from-time :" + fromDate + " to-date :" + ":" + toDate + " filter :" + filter.toString()); Set<Tuple> tupleSet = new HashSet<Tuple>(); LOG.info("Connecting to DB ..."); Connection connection = getConnection( config.getString(AuditDBConstants.JDBC_DRIVER_CLASS_NAME), config.getString(AuditDBConstants.DB_URL), config.getString(AuditDBConstants.DB_USERNAME), config.getString(AuditDBConstants.DB_PASSWORD)); if (connection == null) { LOG.error("Connection not initialized returning ..."); return null; } LOG.info("Connected to DB"); ResultSet rs = null; String statement = getSelectStmtForRetrieve(filter, groupBy); LOG.debug("Select statement :" + statement); PreparedStatement preparedstatement = null; try { preparedstatement = connection.prepareStatement(statement); int index = 1; preparedstatement.setLong(index++, fromDate.getTime()); preparedstatement.setLong(index++, toDate.getTime()); if (filter.getFilters() != null) { for (Column column : Column.values()) { List<String> values = filter.getFilters().get(column); if (values != null && !values.isEmpty()) { for (String value : values) { preparedstatement.setString(index++, value); } } } } LOG.debug("Prepared statement is " + preparedstatement.toString()); rs = preparedstatement.executeQuery(); while (rs.next()) { Tuple tuple = createNewTuple(rs, groupBy); if (tuple == null) { LOG.error("Returned null tuple..returning"); return null; } tupleSet.add(tuple); } connection.commit(); } catch (SQLException e) { logNextException("SQLException encountered", e); } finally { try { if (rs != null) rs.close(); if (preparedstatement != null) preparedstatement.close(); connection.close(); } catch (SQLException e) { logNextException("Exception while closing ", e); } } return tupleSet; } AuditDBHelper(ClientConfig config); String getTableName(); static Connection getConnection(ClientConfig config); static Connection getConnection(String driverName, String url,
      String username, String password); boolean update(Set<Tuple> tupleSet); String getSelectStmtForUpdation(); Set<Tuple> retrieve(Date toDate, Date fromDate, Filter filter,
      GroupBy groupBy); String getSelectStmtForRetrieve(Filter filter, GroupBy groupBy); static void logNextException(String message, SQLException e); static Date addDaysToCurrentDate(Integer dayIncrement); static Date addDaysToGivenDate(Date date, int increment); static Long getFirstMilliOfDay(Date date); static final SimpleDateFormat DAY_CHK_FORMATTER; static final SimpleDateFormat TABLE_DATE_FORMATTER; }
@Test(priority = 4) public void testHostnameNull() { GroupBy groupBy = new GroupBy("CLUSTER,TIER,TOPIC"); Filter filter = new Filter("cluster="+tuple1.getCluster()); AuditDBHelper helper = new AuditDBHelper( ClientConfig.loadFromClasspath(AuditDBConstants.FEEDER_CONF_FILE)); Set<Tuple> tupleSet = helper.retrieve(toDate, fromDate, filter, groupBy); Assert.assertEquals(2, tupleSet.size()); Iterator<Tuple> tupleSetIter = tupleSet.iterator(); Assert.assertTrue(tupleSetIter.hasNext()); Tuple returnedTuple = tupleSetIter.next(); Assert.assertEquals(null, returnedTuple.getHostname()); returnedTuple = tupleSetIter.next(); Assert.assertEquals(null, returnedTuple.getHostname()); }
@Test(priority = 2) public void testRetrieve() { GroupBy groupBy = new GroupBy("TIER,HOSTNAME,CLUSTER"); Filter filter = new Filter("hostname="+tuple1.getHostname()); AuditDBHelper helper = new AuditDBHelper( ClientConfig.loadFromClasspath(AuditDBConstants.FEEDER_CONF_FILE)); Set<Tuple> tupleSet = helper.retrieve(toDate, fromDate, filter, groupBy); Assert.assertEquals(1, tupleSet.size()); Iterator<Tuple> tupleSetIter = tupleSet.iterator(); Assert.assertTrue(tupleSetIter.hasNext()); Tuple returnedTuple = tupleSetIter.next(); Assert.assertEquals(tuple1.getHostname(), returnedTuple.getHostname()); Assert.assertEquals(tuple1.getTier(), returnedTuple.getTier()); Assert.assertEquals(null, returnedTuple.getTopic()); Assert.assertEquals(tuple1.getSent() + tuple2.getSent() + tuple3.getSent(), returnedTuple.getSent()); for (LatencyColumns latencyColumns : LatencyColumns.values()) { Long val1 = tuple1.getLatencyCountMap().get(latencyColumns); if (val1 == null) val1 = 0l; Long val2 = tuple2.getLatencyCountMap().get(latencyColumns); if (val2 == null) val2 = 0l; Long val3 = tuple3.getLatencyCountMap().get(latencyColumns); if (val3 == null) val3 = 0l; Long val4 = returnedTuple.getLatencyCountMap().get(latencyColumns); if (val4 == null) val4 = 0l; Long valx = val1 + val2 + val3; Assert.assertEquals(valx, val4); } Assert.assertEquals((Long) (tuple1.getLostCount() + tuple2.getLostCount() + tuple3.getLostCount()), returnedTuple.getLostCount()); filter = new Filter(null); tupleSet = helper.retrieve(toDate, fromDate, filter, groupBy); Assert.assertEquals(2, tupleSet.size()); }
@Test(priority = 3) public void testTuplesOrder() { GroupBy groupBy = new GroupBy("CLUSTER,TIER,HOSTNAME,TOPIC"); Filter filter = new Filter("hostname="+tuple1.getHostname()); AuditDBHelper helper = new AuditDBHelper( ClientConfig.loadFromClasspath(AuditDBConstants.FEEDER_CONF_FILE)); Set<Tuple> tupleSet = helper.retrieve(toDate, fromDate, filter, groupBy); Assert.assertEquals(2, tupleSet.size()); Iterator<Tuple> tupleSetIter = tupleSet.iterator(); Assert.assertTrue(tupleSetIter.hasNext()); Tuple returnedTuple = tupleSetIter.next(); Assert.assertEquals(tuple1.getTopic(), returnedTuple.getTopic()); returnedTuple = tupleSetIter.next(); Assert.assertEquals(tuple3.getTopic(), returnedTuple.getTopic()); }