code: def consume_from_kafka(self, topics: List[str]):
      self.consumer.subscribe(topics=topics)
      print("Consuming from Kafka started")
      print(
          "Available topics to consume: ", self.consumer.subscription()
      )
      while True:
          try:
              # SIGINT can't be handled when polling, limit timeout to 1 second.
              msg = self.consumer.poll(1.0)
              if msg is None or msg == {}:
                  continue
              for msg_key, msg_values in msg.items():
                  for msg_val in msg_values:
                      print(
                          f"Key:{msg_val.key}-type({type(msg_val.key)}), "
                          f"Value:{msg_val.value}-type({type(msg_val.value)})"
                      )
          except KeyboardInterrupt:
              break
  
      self.consumer.close()
  
imports: import argparse
  from typing import Dict, List
  from kafka import KafkaConsumer
  from settings import BOOTSTRAP_SERVERS, CONSUME_TOPIC_RIDES_CSV
label_desc: Use image node_id 2 for calling the RideCSVConsumer obj's underlying member method code with example usage: obj.consume_from_kafka(topics) without return types