From 97a1f83851347df3f51eafd3d87476ff63682214 Mon Sep 17 00:00:00 2001
From: Frederik Vanrenterghem <frederik@vanrenterghem.io>
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.39.5