@Test public void testInsertTSDRLogRecordFailure() throws InterruptedException, ExecutionException, TimeoutException { String failureMessage = "Failed to store record"; doReturn(RpcResultBuilder.failed().withError(ErrorType.APPLICATION, failureMessage).buildFuture()) .when(logDataService).storeTSDRLogRecord(any()); final ListenableFuture<RpcResult<InsertTSDRLogRecordOutput>> resultFuture = impl.insertTSDRLogRecord( new InsertTSDRLogRecordInputBuilder().setCollectorCodeName("Test") .setTSDRLogRecord(Collections.singletonList(createTSDRLogRecord("Hello World"))).build()); final RpcResult<InsertTSDRLogRecordOutput> 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<InsertTSDRLogRecordOutput>> insertTSDRLogRecord(InsertTSDRLogRecordInput input) { List<TSDRLogRecord> records = input.getTSDRLogRecord().stream().map(inputRec -> new TSDRLogRecordBuilder().setNodeID(inputRec.getNodeID()) .setRecordAttributes(inputRec.getRecordAttributes()) .setRecordFullText(inputRec.getRecordFullText()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setIndex(inputRec.getIndex()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRLogRecordInput tsdrServiceInput = new StoreTSDRLogRecordInputBuilder().setTSDRLogRecord(records).build(); return Futures.transform(logDataService.storeTSDRLogRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRLogRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRLogRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); }
CollectorSPIImpl implements TsdrCollectorSpiService { @Override public ListenableFuture<RpcResult<InsertTSDRLogRecordOutput>> insertTSDRLogRecord(InsertTSDRLogRecordInput input) { List<TSDRLogRecord> records = input.getTSDRLogRecord().stream().map(inputRec -> new TSDRLogRecordBuilder().setNodeID(inputRec.getNodeID()) .setRecordAttributes(inputRec.getRecordAttributes()) .setRecordFullText(inputRec.getRecordFullText()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setIndex(inputRec.getIndex()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRLogRecordInput tsdrServiceInput = new StoreTSDRLogRecordInputBuilder().setTSDRLogRecord(records).build(); return Futures.transform(logDataService.storeTSDRLogRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRLogRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRLogRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); } }
CollectorSPIImpl implements TsdrCollectorSpiService { @Override public ListenableFuture<RpcResult<InsertTSDRLogRecordOutput>> insertTSDRLogRecord(InsertTSDRLogRecordInput input) { List<TSDRLogRecord> records = input.getTSDRLogRecord().stream().map(inputRec -> new TSDRLogRecordBuilder().setNodeID(inputRec.getNodeID()) .setRecordAttributes(inputRec.getRecordAttributes()) .setRecordFullText(inputRec.getRecordFullText()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setIndex(inputRec.getIndex()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRLogRecordInput tsdrServiceInput = new StoreTSDRLogRecordInputBuilder().setTSDRLogRecord(records).build(); return Futures.transform(logDataService.storeTSDRLogRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRLogRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRLogRecordOutput>failed() .withRpcErrors(result.getErrors()).build(); }, MoreExecutors.directExecutor()); } CollectorSPIImpl(TsdrMetricDataService metricService,TsdrLogDataService logService); }
CollectorSPIImpl implements TsdrCollectorSpiService { @Override public ListenableFuture<RpcResult<InsertTSDRLogRecordOutput>> insertTSDRLogRecord(InsertTSDRLogRecordInput input) { List<TSDRLogRecord> records = input.getTSDRLogRecord().stream().map(inputRec -> new TSDRLogRecordBuilder().setNodeID(inputRec.getNodeID()) .setRecordAttributes(inputRec.getRecordAttributes()) .setRecordFullText(inputRec.getRecordFullText()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setIndex(inputRec.getIndex()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRLogRecordInput tsdrServiceInput = new StoreTSDRLogRecordInputBuilder().setTSDRLogRecord(records).build(); return Futures.transform(logDataService.storeTSDRLogRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRLogRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRLogRecordOutput>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<InsertTSDRLogRecordOutput>> insertTSDRLogRecord(InsertTSDRLogRecordInput input) { List<TSDRLogRecord> records = input.getTSDRLogRecord().stream().map(inputRec -> new TSDRLogRecordBuilder().setNodeID(inputRec.getNodeID()) .setRecordAttributes(inputRec.getRecordAttributes()) .setRecordFullText(inputRec.getRecordFullText()) .setRecordKeys(inputRec.getRecordKeys()) .setTimeStamp(inputRec.getTimeStamp()) .setIndex(inputRec.getIndex()) .setTSDRDataCategory(inputRec.getTSDRDataCategory()).build()).collect(Collectors.toList()); StoreTSDRLogRecordInput tsdrServiceInput = new StoreTSDRLogRecordInputBuilder().setTSDRLogRecord(records).build(); return Futures.transform(logDataService.storeTSDRLogRecord(tsdrServiceInput), result -> { return result.isSuccessful() ? RpcResultBuilder.success(new InsertTSDRLogRecordOutputBuilder().build()) .build() : RpcResultBuilder.<InsertTSDRLogRecordOutput>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);  }