From: Frederik Vanrenterghem Date: Fri, 14 Sep 2018 12:52:23 +0000 (+0800) Subject: Add ResetKafkaConsumerOffset function. X-Git-Url: http://git.vanrenterghem.biz/R/project-using-kafka-in-R.git/commitdiff_plain/97a1f83851347df3f51eafd3d87476ff63682214?hp=5b2ed10b2cd8f0326a38e9658a3d5cecf2849c6e Add ResetKafkaConsumerOffset function. Add barebones R function to set a Kafka consumer's topic offset back to 0. --- 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