TreasureDataSender implements Closeable, Sender { public void send(String dbAndTableTag, ByteBuffer dataBuffer) throws IOException { String[] dbAndTable = dbAndTableTag.split("\\."); String database = dbAndTable[0]; String table = dbAndTable[1]; File file = File.createTempFile("tmp-fluency-", ".msgpack.gz"); try { try (InputStream in = new ByteBufferBackedInputStream(dataBuffer); OutputStream out = new GZIPOutputStream( Files.newOutputStream( file.toPath(), StandardOpenOption.WRITE))) { copyStreams(in, out); } String uniqueId = UUID.randomUUID().toString(); Failsafe.with(retryPolicy).run(() -> importData(database, table, uniqueId, file)); } finally { if (!file.delete()) { LOG.warn("Failed to delete a temp file: {}", file.getAbsolutePath()); } } } TreasureDataSender();  TreasureDataSender(Config config); TDClient getClient(); int getRetryInternalMs(); int getMaxRetryInternalMs(); float getRetryFactor(); int getRetryMax(); int getWorkBufSize(); void send(String dbAndTableTag, ByteBuffer dataBuffer); @Override void close();  }
@Test void send() throws IOException { doAnswer(invocation -> { assertImportedFile(invocation.getArgument(2)); return null; }).when(client).importFile(anyString(), anyString(), any(File.class), anyString()); sender.send(DB_AND_TABLE, ByteBuffer.wrap(DATA)); ArgumentCaptor<String> uniqueIdArgumentCaptor = ArgumentCaptor.forClass(String.class); verify(client, times(1)).importFile( eq(DB), eq(TABLE), any(File.class), uniqueIdArgumentCaptor.capture()); verify(client, times(0)).createDatabase(anyString()); verify(client, times(0)).createTable(anyString(), anyString()); UUID.fromString(uniqueIdArgumentCaptor.getValue()); }