Add comments and variables for clarity.

diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index a2ca952cc03dc55d75a77dfc600065973fb5e261..4a77beb3037cab472757235da05d4948362a327f 100644 (file)
--- a/kafkaConsumer.R
+++ b/kafkaConsumer.R
@@ -7,28 +7,39 @@ library(jsonlite)
 library(dplyr)
 
 kafka_rest_proxy <- "http://localhost:8082"
 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
 
 
 # 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"),
      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"))
 
 fromJSON(content(response, "text"))
 

+# Subscribe it to topic

 response <- POST(url=paste(kafka_rest_proxy,
                            "consumers",
 response <- POST(url=paste(kafka_rest_proxy,
                            "consumers",

-                           "my_json_consumer",
+                           consumer,

                            "instances",
                            "instances",

-                           "my_consumer_instance",
+                           consumer_instance,

                            "subscription", sep="/"),
                content_type("application/vnd.kafka.v2+json"),
                            "subscription", sep="/"),
                content_type("application/vnd.kafka.v2+json"),

-               body = '{"topics":["accesslogapache"]}')
+               body = paste0('{"topics":["',
+                             topic,
+                             '"]}')
+               )

 response
 
 response
 

-# Obtain all messages on the topic
-messagesJSON <- 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")
 
                 accept("application/vnd.kafka.json.v2+json"),
                 encode="json")
 


@@ -38,7 +49,8 @@ createPlot(messages$value)
 
 Sys.sleep(120)
 
 
 Sys.sleep(120)
 

-messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance","records", sep = "/"),
+# 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")
 
                 accept("application/vnd.kafka.json.v2+json"),
                 encode="json")
 


@@ -49,5 +61,5 @@ apachelog <- rbind(messages$value,messages2$value)
 createPlot(apachelog)
 
 # Remove the consumer
 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"))
        content_type("application/vnd.kafka.v2+json"))