static KafkaSupervisorSpec createKafkaSupervisorSpec(
      Table table,
      String kafkaTopic,
      String kafkaServers,
      DataSchema dataSchema,
      IndexSpec indexSpec
  ) {
    return new KafkaSupervisorSpec(dataSchema,
        new KafkaSupervisorTuningConfig(DruidStorageHandlerUtils.getIntegerProperty(table,
            DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "maxRowsInMemory"),
                DruidStorageHandlerUtils.getLongProperty(table,
                        DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "maxBytesInMemory"),

            DruidStorageHandlerUtils.getIntegerProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "maxRowsPerSegment"), DruidStorageHandlerUtils
            .getLongProperty(table, DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "maxTotalRows"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "intermediatePersistPeriod"), null,
            // basePersistDirectory - use druid default, no need to be configured by user
            DruidStorageHandlerUtils
                .getIntegerProperty(table, DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "maxPendingPersists"),
            indexSpec, null, null,
            // buildV9Directly - use druid default, no need to be configured by user
            DruidStorageHandlerUtils.getBooleanProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "reportParseExceptions"),
            DruidStorageHandlerUtils.getLongProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "handoffConditionTimeout"),
            DruidStorageHandlerUtils.getBooleanProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "resetOffsetAutomatically"),
            TmpFileSegmentWriteOutMediumFactory.instance(), DruidStorageHandlerUtils
            .getIntegerProperty(table, DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "workerThreads"),
            DruidStorageHandlerUtils.getIntegerProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "chatThreads"),
            DruidStorageHandlerUtils.getLongProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "chatRetries"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "httpTimeout"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "shutdownTimeout"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "offsetFetchPeriod"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                    DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "intermediateHandoffPeriod"),
    DruidStorageHandlerUtils.getBooleanProperty(table,
            DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "logParseExceptions"),
    DruidStorageHandlerUtils.getIntegerProperty(table,
            DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "maxParseExceptions"),
    DruidStorageHandlerUtils.getIntegerProperty(table,
            DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "maxSavedParseExceptions")),
        new KafkaSupervisorIOConfig(kafkaTopic,
            // Mandatory Property
            DruidStorageHandlerUtils.getIntegerProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "replicas"),
            DruidStorageHandlerUtils.getIntegerProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "taskCount"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "taskDuration"),
            getKafkaConsumerProperties(table, kafkaServers),
            // Mandatory Property
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "startDelay"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "period"),
            DruidStorageHandlerUtils.getBooleanProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "useEarliestOffset"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "completionTimeout"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "lateMessageRejectionPeriod"),
            DruidStorageHandlerUtils.getPeriodProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "earlyMessageRejectionPeriod"),
            DruidStorageHandlerUtils.getBooleanProperty(table,
                DruidConstants.DRUID_KAFKA_INGESTION_PROPERTY_PREFIX + "skipOffsetGaps")),
        new HashMap<>());
  }
Class containing some Utility methods for Kafka Ingestion.
druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java