From 97a1f83851347df3f51eafd3d87476ff63682214 Mon Sep 17 00:00:00 2001 From: Frederik Vanrenterghem Date: Fri, 14 Sep 2018 20:52:23 +0800 Subject: [PATCH] Add ResetKafkaConsumerOffset function. Add barebones R function to set a Kafka consumer's topic offset back to 0. --- R/ResetKafkaConsumerOffset.R | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 R/ResetKafkaConsumerOffset.R diff --git a/R/ResetKafkaConsumerOffset.R b/R/ResetKafkaConsumerOffset.R new file mode 100644 index 0000000..7383fef --- /dev/null +++ b/R/ResetKafkaConsumerOffset.R @@ -0,0 +1,13 @@ +ResetKafkaConsumerOffset <- function(consumer.base.uri, topic) { + response <- POST(url=paste(consumer.base.uri, + "positions", "beginning", sep="/"), + content_type("application/vnd.kafka.v2+json"), + body = paste0('{"partitions": [ {"topic": "', + topic, + '" , "partition": 0}', + ']}') + ) + if(response$status_code == 204) { + return(response) + } else stop(response) +} \ No newline at end of file -- 2.30.2