static void updateKafkaIngestionSpec(String overlordAddress, KafkaSupervisorSpec spec) {
    try {
      String task = JSON_MAPPER.writeValueAsString(spec);
      CONSOLE.printInfo("submitting kafka Spec {}", task);
      LOG.info("submitting kafka Supervisor Spec {}", task);
      StringFullResponseHolder response = DruidStorageHandlerUtils
          .getResponseFromCurrentLeader(DruidStorageHandler.getHttpClient(), new Request(HttpMethod.POST,
                  new URL(String.format("http://%s/druid/indexer/v1/supervisor", overlordAddress)))
                  .setContent("application/json", JSON_MAPPER.writeValueAsBytes(spec)),
              new StringFullResponseHandler(Charset.forName("UTF-8")));
      if (response.getStatus().equals(HttpResponseStatus.OK)) {
        String
            msg =
            String.format("Kafka Supervisor for [%s] Submitted Successfully to druid.",
                spec.getDataSchema().getDataSource());
        LOG.info(msg);
        CONSOLE.printInfo(msg);
      } else {
        throw new IOException(String.format("Unable to update Kafka Ingestion for Druid status [%d] full response [%s]",
            response.getStatus().getCode(),
            response.getContent()));
      }
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }
Class containing some Utility methods for Kafka Ingestion.
druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java