MetricCollectorResource { @POST @Consumes({MediaType.APPLICATION_JSON}) public Response doPost( List<Map<String, Object>> events, @QueryParam(_CLUSTER) final String cluster ) throws IOException { if (Strings.isNullOrEmpty(cluster)) { return Response.status(Response.Status.BAD_REQUEST) .entity(String.format("request must have %s query param", _CLUSTER)).build(); } if (events == null || events.isEmpty()) { return Response.status(Response.Status.BAD_REQUEST) .entity("request must have one or more metric events").build(); } final Callback callback = new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { eventLogger.countAsSent(cluster); } else { eventLogger.countAsLost(cluster); log.error( exception, "Error to send kafka record from cluster [%s], error message [%s] ", cluster, exception.getMessage() ); } } }; for (final Map<String, Object> event : events) { Object feedName = event.get(METRIC_FEED_KEY); if (feedName == null || !VALID_FEED_NAMES.contains(feedName)) { return Response.status(Response.Status.BAD_REQUEST) .entity("Each metric event must have a valid feed name").build(); } event.put(_CLUSTER, cluster); final String topic = kafkaTopicPrefix + "-" + feedName; final String eventString = jsonMapper.writeValueAsString(event); eventLogger.incrementEventSize(cluster, eventString.getBytes().length); final ProducerRecord<String, String> record = new ProducerRecord( topic, null, eventString ); try { producer.send(record, callback); } catch (Exception exception) { eventLogger.countAsLost(cluster); log.error(exception, "Got Exception"); return Response.serverError().entity("Something went wrong in the POST request, emitter will retry !").build(); } } return Response.ok().build(); } @Inject  MetricCollectorResource(
      ObjectMapper jsonMapper,
      MetricCollectorConfig config,
      @MetricCollector Producer<String, String> producer,
      ServiceEmitter emitter
  ); @LifecycleStart void start(); @LifecycleStop void stop(); @POST @Consumes({MediaType.APPLICATION_JSON}) Response doPost(
      List<Map<String, Object>> events,
      @QueryParam(_CLUSTER) final String cluster
  );  }
@Test public void testDoPostNoEvents() throws IOException { Response resp = resource.doPost(Lists.<Map<String, Object>>newArrayList(), TEST_CLUSTER); Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), resp.getStatus()); }
@Test public void testDoPostNoCluster() throws IOException { List<Map<String, Object>> in = buildTestEvent("metrics"); Response resp = resource.doPost(in, ""); Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), resp.getStatus()); }
@Test public void testDoPostInvalidFeed() throws IOException { List<Map<String, Object>> in = buildTestEvent("xxx"); Response resp = resource.doPost(in, TEST_CLUSTER); Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), resp.getStatus()); }
@Test public void testDoPost() throws IOException { Capture<ProducerRecord<String, String>> capturedMsg = new Capture<ProducerRecord<String, String>>(); EasyMock.expect( producer.send( EasyMock.<ProducerRecord<String, String>>capture(capturedMsg), EasyMock.anyObject(Callback.class) ) ).andReturn(null); EasyMock.replay(producer); List<Map<String, Object>> in = buildTestEvent("metrics"); Response resp = resource.doPost(in, TEST_CLUSTER); Assert.assertEquals(Response.Status.OK.getStatusCode(), resp.getStatus()); ProducerRecord<String, String> msg = capturedMsg.getValue(); List<Map<String, Object>> actualEvents = Lists.newArrayList(); Map<String, Object> event = jsonMapper.readValue( msg.value(), new TypeReference<Map<String, Object>>() { } ); Assert.assertEquals(KAFKA_TOPIC_PREFIX + "-" + event.get(METRIC_FEED_KEY), msg.topic()); Assert.assertEquals(TEST_CLUSTER, event.get(_CLUSTER)); actualEvents.add(event); Assert.assertEquals(in, actualEvents); }