HttpElasticIndexWriter implements ElasticIndexWriter { @Override public void write(final String idFieldName, final String index, final String type, final Map<String, String> values, final Consumer<Exception> exceptionHandler) { HttpURLConnection con = null; try { String recordId = UUID.randomUUID().toString(); if (null != idFieldName) { String idFieldValue = values.get(idFieldName); if (null != idFieldValue) { recordId = idFieldValue; } } final String url = String.format("%s/%s/%s/%s", this.elasticHttpUrl, index, type, recordId); URL obj = new URL(url); con = (HttpURLConnection) obj.openConnection(); con.setRequestMethod("PUT"); con.setRequestProperty("Accept-Language", "en-US,en;q=0.5"); con.setRequestProperty("Content-Type", "application/json"); String body = jsonMapper.writeValueAsString(values); con.setDoOutput(true); DataOutputStream wr = new DataOutputStream(con.getOutputStream()); wr.writeBytes(body); wr.flush(); wr.close(); int responseCode = con.getResponseCode(); BufferedReader in = new BufferedReader( new InputStreamReader(con.getInputStream())); String inputLine; StringBuilder response = new StringBuilder(); while ((inputLine = in.readLine()) != null) { response.append(inputLine); } in.close(); switch (responseCode) { case HttpURLConnection.HTTP_CREATED: case HttpURLConnection.HTTP_ACCEPTED: case HttpURLConnection.HTTP_OK: break; default: final String msg = String.format("Bad Response from Elastic Search: %d - %s", responseCode, response); exceptionHandler.accept(new Exception(msg)); break; } } catch (final IOException e) { final String msg = String.format("Error initialising elastic producer to %s, (%s)", this.elasticHttpUrl, e.getMessage()); LOGGER.error(msg); exceptionHandler.accept(e); } finally { if (null != con) { con.disconnect(); } } } HttpElasticIndexWriter(final ElasticIndexConfigDoc elasticIndexConfigDoc); @Override void write(final String idFieldName,
                      final String index,
                      final String type,
                      final Map<String, String> values,
                      final Consumer<Exception> exceptionHandler);  }
@Test @Disabled("You may use this to test the local instance of Elastic.") void testManualSend() { final String recordId = UUID.randomUUID().toString(); final ElasticIndexConfigDoc elasticIndexConfigDoc = new ElasticIndexConfigDoc(); elasticIndexConfigDoc.setElasticHttpUrl("http: final ElasticIndexWriter stroomElasticProducer = new HttpElasticIndexWriter(elasticIndexConfigDoc); final String STAT_NAME = "stat-name"; final String STAT_TYPE = "stat-type"; final String STAT_TIME = "stat-time"; final Map<String, String> record = new HashMap<>(); record.put(STAT_NAME, UUID.randomUUID().toString()); record.put(STAT_TYPE, "test"); record.put(STAT_TIME, new Date().toString()); stroomElasticProducer.write(recordId, "statistics", "line", record, DEFAULT_CALLBACK); }