Migrate the hub codebase to use the latest Kafka connector records

Author: ayeshLK

examples/kafka-hub/consolidator/consolidator_service.bal

@@ -37,8 +37,8 @@ http:Service consolidatorService = service object {
 isolated function consolidateSystemState() returns error? {
     do {
         while true {
-            kafka:ConsumerRecord[] records = check conn:websubEventConsumer->poll(config:POLLING_INTERVAL);
-            foreach kafka:ConsumerRecord currentRecord in records {
+            kafka:BytesConsumerRecord[] records = check conn:websubEventConsumer->poll(config:POLLING_INTERVAL);
+            foreach kafka:BytesConsumerRecord currentRecord in records {
                 string lastPersistedData = check string:fromBytes(currentRecord.value);
                 error? result = processPersistedData(lastPersistedData);
                 if result is error {

examples/kafka-hub/hub/hub_state_update.bal

@@ -39,11 +39,11 @@ function initializeHubState() returns error? {
 
 function updateHubState() returns error? {
     while true {
-        kafka:ConsumerRecord[] records = check conn:websubEventsConsumer->poll(config:POLLING_INTERVAL);
+        kafka:BytesConsumerRecord[] records = check conn:websubEventsConsumer->poll(config:POLLING_INTERVAL);
         if records.length() <= 0 {
             continue;
         }
-        foreach kafka:ConsumerRecord currentRecord in records {
+        foreach kafka:BytesConsumerRecord currentRecord in records {
             string lastPersistedData = check string:fromBytes(currentRecord.value);
             error? result = processStateUpdateEvent(lastPersistedData);
             if result is error {

examples/kafka-hub/hub/websub_subscribers.bal

@@ -85,7 +85,7 @@ isolated function pollForNewUpdates(string subscriberId, websubhub:VerifiedSubsc
     });
     do {
         while true {
-            kafka:ConsumerRecord[] records = check consumerEp->poll(config:POLLING_INTERVAL);
+            kafka:BytesConsumerRecord[] records = check consumerEp->poll(config:POLLING_INTERVAL);
             if !isValidConsumer(topicName, subscriberId) {
                 fail error(string `Subscriber with Id ${subscriberId} or topic ${topicName} is invalid`);
             }
@@ -113,7 +113,7 @@ isolated function isValidSubscription(string subscriberId) returns boolean {
     }
 }
 
-isolated function notifySubscribers(kafka:ConsumerRecord[] records, websubhub:HubClient clientEp, kafka:Consumer consumerEp) returns error? {
+isolated function notifySubscribers(kafka:BytesConsumerRecord[] records, websubhub:HubClient clientEp, kafka:Consumer consumerEp) returns error? {
     foreach var kafkaRecord in records {
         var message = deSerializeKafkaRecord(kafkaRecord);
         if message is websubhub:ContentDistributionMessage {
@@ -129,7 +129,7 @@ isolated function notifySubscribers(kafka:ConsumerRecord[] records, websubhub:Hu
     }
 }
 
-isolated function deSerializeKafkaRecord(kafka:ConsumerRecord kafkaRecord) returns websubhub:ContentDistributionMessage|error {
+isolated function deSerializeKafkaRecord(kafka:BytesConsumerRecord kafkaRecord) returns websubhub:ContentDistributionMessage|error {
     byte[] content = kafkaRecord.value;
     string message = check string:fromBytes(content);
     json payload =  check value:fromJsonString(message);