@Test public void testInsertTSDRMetricRecordFailure() throws InterruptedException, ExecutionException, TimeoutException { String failureMessage = "Failed to store record"; doReturn(RpcResultBuilder.failed().withError(ErrorType.APPLICATION, failureMessage).buildFuture()) .when(metricDataService).storeTSDRMetricRecord(any()); final ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> resultFuture = impl.insertTSDRMetricRecord( new InsertTSDRMetricRecordInputBuilder().setCollectorCodeName("Test") .setTSDRMetricRecord(Collections.singletonList( createTSDRMetricRecord("Test Metric", BigDecimal.valueOf(123)))).build()); final RpcResult<InsertTSDRMetricRecordOutput> result = resultFuture.get(5, TimeUnit.SECONDS); assertFalse(result.isSuccessful()); assertEquals(1, result.getErrors().size()); assertEquals(failureMessage, result.getErrors().iterator().next().getMessage()); assertEquals(ErrorType.APPLICATION, result.getErrors().iterator().next().getErrorType()); }
@Override public ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> insertTSDRMetricRecord( InsertTSDRMetricRecordInput input) { List<TSDRMetricRecord> records = input.getTSDRMetricRecord().stream().map(inputRec -> new TSDRMetricRecordBuilder().setMetricName(inputRec.getMetricName()) .setMetricValue(inputRec.getMetricValue()) .setNodeID(inputRec.getNodeID()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRMetricRecordInput tsdrServiceInput = new StoreTSDRMetricRecordInputBuilder().setTSDRMetricRecord(records).build(); return Futures.transform(metricDataService.storeTSDRMetricRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRMetricRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRMetricRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); }
CollectorSPIImpl implements TsdrCollectorSpiService { @Override public ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> insertTSDRMetricRecord( InsertTSDRMetricRecordInput input) { List<TSDRMetricRecord> records = input.getTSDRMetricRecord().stream().map(inputRec -> new TSDRMetricRecordBuilder().setMetricName(inputRec.getMetricName()) .setMetricValue(inputRec.getMetricValue()) .setNodeID(inputRec.getNodeID()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRMetricRecordInput tsdrServiceInput = new StoreTSDRMetricRecordInputBuilder().setTSDRMetricRecord(records).build(); return Futures.transform(metricDataService.storeTSDRMetricRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRMetricRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRMetricRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); } }
CollectorSPIImpl implements TsdrCollectorSpiService { @Override public ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> insertTSDRMetricRecord( InsertTSDRMetricRecordInput input) { List<TSDRMetricRecord> records = input.getTSDRMetricRecord().stream().map(inputRec -> new TSDRMetricRecordBuilder().setMetricName(inputRec.getMetricName()) .setMetricValue(inputRec.getMetricValue()) .setNodeID(inputRec.getNodeID()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRMetricRecordInput tsdrServiceInput = new StoreTSDRMetricRecordInputBuilder().setTSDRMetricRecord(records).build(); return Futures.transform(metricDataService.storeTSDRMetricRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRMetricRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRMetricRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); } CollectorSPIImpl(TsdrMetricDataService metricService,TsdrLogDataService logService); }
CollectorSPIImpl implements TsdrCollectorSpiService { @Override public ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> insertTSDRMetricRecord( InsertTSDRMetricRecordInput input) { List<TSDRMetricRecord> records = input.getTSDRMetricRecord().stream().map(inputRec -> new TSDRMetricRecordBuilder().setMetricName(inputRec.getMetricName()) .setMetricValue(inputRec.getMetricValue()) .setNodeID(inputRec.getNodeID()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRMetricRecordInput tsdrServiceInput = new StoreTSDRMetricRecordInputBuilder().setTSDRMetricRecord(records).build(); return Futures.transform(metricDataService.storeTSDRMetricRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRMetricRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRMetricRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); } CollectorSPIImpl(TsdrMetricDataService metricService,TsdrLogDataService logService); @Override ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> insertTSDRMetricRecord(
            InsertTSDRMetricRecordInput input); @Override ListenableFuture<RpcResult<InsertTSDRLogRecordOutput>> insertTSDRLogRecord(InsertTSDRLogRecordInput input); }
CollectorSPIImpl implements TsdrCollectorSpiService { @Override public ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> insertTSDRMetricRecord( InsertTSDRMetricRecordInput input) { List<TSDRMetricRecord> records = input.getTSDRMetricRecord().stream().map(inputRec -> new TSDRMetricRecordBuilder().setMetricName(inputRec.getMetricName()) .setMetricValue(inputRec.getMetricValue()) .setNodeID(inputRec.getNodeID()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRMetricRecordInput tsdrServiceInput = new StoreTSDRMetricRecordInputBuilder().setTSDRMetricRecord(records).build(); return Futures.transform(metricDataService.storeTSDRMetricRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRMetricRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRMetricRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); } CollectorSPIImpl(TsdrMetricDataService metricService,TsdrLogDataService logService); @Override ListenableFuture<RpcResult<InsertTSDRMetricRecordOutput>> insertTSDRMetricRecord(
            InsertTSDRMetricRecordInput input); @Override ListenableFuture<RpcResult<InsertTSDRLogRecordOutput>> insertTSDRLogRecord(InsertTSDRLogRecordInput input);  }