./bin/kafka-server-start.sh config/server.properties

./bin/kafka-topics.sh --create --zookeeper {kafka_zk_ip}:2181 --replication-factor 1 --partitions 1 --topic mytopic
Output:
Created topic "mytopic".

{
  "type": "kafka",
  "dataSchema": {
      "dataSource": "mytopic-kafka",
      "parser": {
          "type": "string",
          "parseSpec": {
              "timestampSpec": {
                  "column": "time",
                  "format": "auto"
              },
              "dimensionsSpec": {
                  "dimensions": ["url", "user"]
              },
              "format": "json"
          }
      },
      "granularitySpec": {
          "type": "uniform",
          "segmentGranularity": "hour",
          "queryGranularity": "none"
      },
      "metricsSpec": [{
              "type": "count",
              "name": "views"
          },
          {
              "name": "latencyMs",
              "type": "doubleSum",
              "fieldName": "latencyMs"
          }
      ]
  },
  "ioConfig": {
      "topic": "mytopic",
      "consumerProperties": {
          "bootstrap.servers": "{kafka_ip}:9092",
          "group.id": "kafka-indexing-service"
      },
      "taskCount": 1,
      "replicas": 1,
      "taskDuration": "PT1H"
  },
  "tuningConfig": {
      "type": "kafka",
      "maxRowsInMemory": "100000"
  }
 }

curl -XPOST -H 'Content-Type: application/json' -d @kafka-mytopic.json http://{druid_master_ip}:8090/druid/indexer/v1/supervisor
Output:
{"id":"mytopic-kafka"}

./bin/kafka-console-producer.sh --broker-list {kafka_ip}:9092 --topic mytopic

{
  "queryType" : "search",
  "dataSource" : "mytopic-kafka",
  "intervals" : ["2020-03-13T00:00:00.000/2020-03-20T00:00:00.000"],
  "granularity" : "all",
  "searchDimensions": [
      "url",
      "user"
  ],
  "query": {
      "type": "insensitive_contains",
      "value": "roni"
  }
 }

{"time": "2020-03-19T09:57:58Z", "url": "/foo/bar", "user": "brozo", "latencyMs": 62}
{"time": "2020-03-19T16:57:59Z", "url": "/", "user": "roni", "latencyMs": 15}
{"time": "2020-03-19T17:50:00Z", "url": "/foo/bar", "user": "roni", "latencyMs": 25}

python -c 'import datetime; print(datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"))'

curl -XPOST -H 'Content-Type: application/json' -d @query-mytopic.json http://{druid_ip}:8082/druid/v2/?pretty

[ {
  "timestamp" : "2020-03-19T16:00:00.000Z",
  "result" : [ {
 "dimension" : "user",
 "value" : "roni",
 "count" : 2
  } ]
} ]