R/SubscribeKafkaTopic.R

   1 #' Subscribe Apache Kafka consumer to a topic using Confluent's REST Proxy API
   2 #'
   3 #' \code{SubscribeKafkaTopic} subscribes an Apache Kafka consumer to a topic
   4 #'   using the rest proxy api provided by Confluent.
   5 #'
   6 #' @param consumer.base.uri The URI of the consumer created by \code{\link{CreateKafkaConsumer}}.
   7 #' @param topic A chr object with the topic to subscribe to.
   8 #'
   9 #' @return If the operation succeeds, returns the response, which can be discarded.
  10 #'   If the operation fails, it might help out understanding what happened.
  11 #'
  12 SubscribeKafkaTopic <- function(consumer.base.uri, topic) {
  13   response <- POST(url=paste(consumer.base.uri,
  14                              "subscription", sep="/"),
  15                    content_type("application/vnd.kafka.v2+json"),
  16                    body = paste0('{"topics":["',
  17                                  topic,
  18                                  '"]}')
  19   )
  20   if(response$status_code == 204) {
  21     return(response)
  22   } else stop(response)
  23 }