@Test @Ignore public void testUploadEvent() throws Exception { Properties props = System.getProperties(); props.load(this.getClass().getClassLoader().getResourceAsStream("treasure-data.properties")); MessagePack msgpack = new MessagePack(); String apiKey = props.getProperty(Config.TD_LOGGER_API_KEY); TreasureDataClient client = new TreasureDataClient( new TreasureDataCredentials(apiKey), props); HttpSenderThread thread = new HttpSenderThread(null, client); ByteArrayOutputStream out = new ByteArrayOutputStream(); GZIPOutputStream gzout = new GZIPOutputStream(out); Packer packer = msgpack.createPacker(gzout); for (int i = 0; i < 20; ++i) { Map<String, Object> data = new HashMap<String, Object>(); data.put("k", i); data.put("time", System.currentTimeMillis() / 1000); packer.write(data); } gzout.finish(); byte[] bytes = out.toByteArray(); System.out.println(bytes.length); QueueEvent ev = new QueueEvent("mugadb", "table01", bytes, 20); thread.uploadEvent(ev); }
void uploadEvent(QueueEvent ev) throws IOException, ClientException { boolean retry = true; while (retry) { try { LOG.info(String.format("Uploading event logs to %s.%s on TreasureData (%d bytes, %d records)", ev.databaseName, ev.tableName, ev.data.length, ev.rowCount)); Table table = new Table(new Database(ev.databaseName), ev.tableName); long time = System.currentTimeMillis(); client.importData(table, ev.data); time = System.currentTimeMillis() - time; if (LOG.isLoggable(Level.FINE)) { LOG.fine(String.format("Uploaded in %d sec.", time)); } retry = false; } catch (ClientException e) { if (!Boolean.parseBoolean(System.getProperty( Config.TD_LOGGER_AUTO_CREATE_TABLE, Config.TD_LOGGER_AUTO_CREATE_TABLE_DEFAULT))) { throw e; } LOG.info(String.format("Creating table %s.%s on Treasure Data", new Object[] { ev.databaseName, ev.tableName })); try { client.createTable(ev.databaseName, ev.tableName); } catch (ClientException e0) { client.createDatabase(ev.databaseName); client.createTable(ev.databaseName, ev.tableName); } } } }
HttpSenderThread implements Runnable { void uploadEvent(QueueEvent ev) throws IOException, ClientException { boolean retry = true; while (retry) { try { LOG.info(String.format("Uploading event logs to %s.%s on TreasureData (%d bytes, %d records)", ev.databaseName, ev.tableName, ev.data.length, ev.rowCount)); Table table = new Table(new Database(ev.databaseName), ev.tableName); long time = System.currentTimeMillis(); client.importData(table, ev.data); time = System.currentTimeMillis() - time; if (LOG.isLoggable(Level.FINE)) { LOG.fine(String.format("Uploaded in %d sec.", time)); } retry = false; } catch (ClientException e) { if (!Boolean.parseBoolean(System.getProperty( Config.TD_LOGGER_AUTO_CREATE_TABLE, Config.TD_LOGGER_AUTO_CREATE_TABLE_DEFAULT))) { throw e; } LOG.info(String.format("Creating table %s.%s on Treasure Data", new Object[] { ev.databaseName, ev.tableName })); try { client.createTable(ev.databaseName, ev.tableName); } catch (ClientException e0) { client.createDatabase(ev.databaseName); client.createTable(ev.databaseName, ev.tableName); } } } } }
HttpSenderThread implements Runnable { void uploadEvent(QueueEvent ev) throws IOException, ClientException { boolean retry = true; while (retry) { try { LOG.info(String.format("Uploading event logs to %s.%s on TreasureData (%d bytes, %d records)", ev.databaseName, ev.tableName, ev.data.length, ev.rowCount)); Table table = new Table(new Database(ev.databaseName), ev.tableName); long time = System.currentTimeMillis(); client.importData(table, ev.data); time = System.currentTimeMillis() - time; if (LOG.isLoggable(Level.FINE)) { LOG.fine(String.format("Uploaded in %d sec.", time)); } retry = false; } catch (ClientException e) { if (!Boolean.parseBoolean(System.getProperty( Config.TD_LOGGER_AUTO_CREATE_TABLE, Config.TD_LOGGER_AUTO_CREATE_TABLE_DEFAULT))) { throw e; } LOG.info(String.format("Creating table %s.%s on Treasure Data", new Object[] { ev.databaseName, ev.tableName })); try { client.createTable(ev.databaseName, ev.tableName); } catch (ClientException e0) { client.createDatabase(ev.databaseName); client.createTable(ev.databaseName, ev.tableName); } } } } HttpSenderThread(HttpSender sender, TreasureDataClient client); }
HttpSenderThread implements Runnable { void uploadEvent(QueueEvent ev) throws IOException, ClientException { boolean retry = true; while (retry) { try { LOG.info(String.format("Uploading event logs to %s.%s on TreasureData (%d bytes, %d records)", ev.databaseName, ev.tableName, ev.data.length, ev.rowCount)); Table table = new Table(new Database(ev.databaseName), ev.tableName); long time = System.currentTimeMillis(); client.importData(table, ev.data); time = System.currentTimeMillis() - time; if (LOG.isLoggable(Level.FINE)) { LOG.fine(String.format("Uploaded in %d sec.", time)); } retry = false; } catch (ClientException e) { if (!Boolean.parseBoolean(System.getProperty( Config.TD_LOGGER_AUTO_CREATE_TABLE, Config.TD_LOGGER_AUTO_CREATE_TABLE_DEFAULT))) { throw e; } LOG.info(String.format("Creating table %s.%s on Treasure Data", new Object[] { ev.databaseName, ev.tableName })); try { client.createTable(ev.databaseName, ev.tableName); } catch (ClientException e0) { client.createDatabase(ev.databaseName); client.createTable(ev.databaseName, ev.tableName); } } } } HttpSenderThread(HttpSender sender, TreasureDataClient client); @Override void run(); }
HttpSenderThread implements Runnable { void uploadEvent(QueueEvent ev) throws IOException, ClientException { boolean retry = true; while (retry) { try { LOG.info(String.format("Uploading event logs to %s.%s on TreasureData (%d bytes, %d records)", ev.databaseName, ev.tableName, ev.data.length, ev.rowCount)); Table table = new Table(new Database(ev.databaseName), ev.tableName); long time = System.currentTimeMillis(); client.importData(table, ev.data); time = System.currentTimeMillis() - time; if (LOG.isLoggable(Level.FINE)) { LOG.fine(String.format("Uploaded in %d sec.", time)); } retry = false; } catch (ClientException e) { if (!Boolean.parseBoolean(System.getProperty( Config.TD_LOGGER_AUTO_CREATE_TABLE, Config.TD_LOGGER_AUTO_CREATE_TABLE_DEFAULT))) { throw e; } LOG.info(String.format("Creating table %s.%s on Treasure Data", new Object[] { ev.databaseName, ev.tableName })); try { client.createTable(ev.databaseName, ev.tableName); } catch (ClientException e0) { client.createDatabase(ev.databaseName); client.createTable(ev.databaseName, ev.tableName); } } } } HttpSenderThread(HttpSender sender, TreasureDataClient client); @Override void run();  }