SqlServerSinkHadoop1 extends SqlServerSink { @Override public void putMetrics(MetricsRecord record) { long metricRecordID = getMetricRecordID(record.context(), record.name(), getLocalNodeName(), getLocalNodeIPAddress(), getClusterNodeName(), getCurrentServiceName(), getTagString(record.tags()), record.timestamp()); if (metricRecordID < 0) return; for (Metric metric : record.metrics()) { insertMetricValue(metricRecordID, metric.name(), String.valueOf(metric .value())); if (metric.name().equals("BlockCapacity")) { insertMetricValue(metricRecordID, "BlockSize", Integer .toString(getBlockSize())); } } } SqlServerSinkHadoop1(); @Override void putMetrics(MetricsRecord record);  }
@Override @Test public void testPutMetrics() throws Exception { SubsetConfiguration configuration = createNiceMock(SubsetConfiguration.class); Connection connection = createNiceMock(Connection.class); CallableStatement cstmt = createNiceMock(CallableStatement.class); MetricsRecord record = createNiceMock(MetricsRecord.class); Metric metric = createNiceMock(Metric.class); expect(configuration.getParent()).andReturn(null); expect(configuration.getPrefix()).andReturn("prefix"); expect(configuration.getString("databaseUrl")).andReturn("url"); expect(record.context()).andReturn("context"); expect(record.name()).andReturn("typeName"); expect(record.tags()).andReturn(new HashSet<MetricsTag>()); expect(record.timestamp()).andReturn(9999L); expect(record.metrics()).andReturn(Collections.singleton(metric)); expect(metric.name()).andReturn("name").anyTimes(); expect(metric.value()).andReturn(1234); expect(connection.prepareCall("{call dbo.uspGetMetricRecord(?, ?, ?, ?, ?, ?, ?, ?, ?)}")).andReturn(cstmt); cstmt.setNString(1, "context"); cstmt.setNString(2, "typeName"); cstmt.setNString(eq(3), (String) anyObject()); cstmt.setNString(eq(4), (String) anyObject()); cstmt.setNString(eq(5), (String) anyObject()); cstmt.setNString(6, "prefix"); cstmt.setNString(7, "sourceName:prefix"); cstmt.setLong(8, 9999L); cstmt.registerOutParameter(9, java.sql.Types.BIGINT); expect(cstmt.execute()).andReturn(true); expect(cstmt.getLong(9)).andReturn(99L); expect(cstmt.wasNull()).andReturn(false); expect(connection.prepareCall("{call dbo.uspInsertMetricValue(?, ?, ?)}")).andReturn(cstmt); cstmt.setLong(1, 99L); cstmt.setNString(2, "name"); cstmt.setNString(3, "1234"); expect(cstmt.execute()).andReturn(true); replay(configuration, connection, cstmt, record, metric); SqlServerSink sink = createInstance(); sink.init(configuration); SQLServerDriver.setConnection(connection); sink.putMetrics(record); verify(configuration, connection, cstmt, record, metric); }