library(httr)
library(jsonlite)
+library(dplyr)
+
kafka_rest_proxy <- "http://localhost:8082"
+consumer <- 'my_json_consumer'
+consumer_instance <- 'my_consumer_instance'
+topic <- 'accesslogapache'
# Create consumer
# See https://docs.confluent.io/current/kafka-rest/docs/intro.html#produce-and-consume-avro-messages
-response <- POST(url=paste(kafka_rest_proxy, "consumers", "my_json_consumer", sep="/"),
+# Create consumer
+response <- POST(url=paste(kafka_rest_proxy, "consumers", consumer, sep="/"),
content_type("application/vnd.kafka.v2+json"),
accept("application/vnd.kafka.v2+json"),
- body='{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}')
+ body=paste0('{"name": "',
+ consumer_instance,
+ '", "format": "json", "auto.offset.reset": "earliest"}')
+ )
fromJSON(content(response, "text"))
+# Subscribe it to topic
response <- POST(url=paste(kafka_rest_proxy,
"consumers",
- "my_json_consumer",
+ consumer,
"instances",
- "my_consumer_instance",
+ consumer_instance,
"subscription", sep="/"),
content_type("application/vnd.kafka.v2+json"),
- body = '{"topics":["accesslogapache"]}')
+ body = paste0('{"topics":["',
+ topic,
+ '"]}')
+ )
response
-# Obtain all messages on the topic
-messages <- GET(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance","records", sep = "/"),
+# Obtain all (or latest) messages on the topic
+messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance,"records", sep = "/"),
+ accept("application/vnd.kafka.json.v2+json"),
+ encode="json")
+
+messages <- fromJSON(content(messagesJSON,"text"))
+
+createPlot(messages$value)
+
+Sys.sleep(120)
+
+# Obtain latest messages
+messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance,"records", sep = "/"),
accept("application/vnd.kafka.json.v2+json"),
encode="json")
-apachelog <- fromJSON(content(messages,"text"))
+messages2 <- fromJSON(content(messagesJSON,"text"))
+
+apachelog <- rbind(messages$value,messages2$value)
+
+createPlot(apachelog)
# Remove the consumer
-DELETE(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance", sep = "/"),
+DELETE(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance, sep = "/"),
content_type("application/vnd.kafka.v2+json"))