class_docstr: Test ConsumeFromTopic
code: def test_operator_consume_max(self, mock_get_consumer):
      mock_consumer = mock.MagicMock()
  
      mocked_messages = ["test_messages" for i in range(1001)]
  
      def mock_consume(num_messages=0, timeout=-1):
          nonlocal mocked_messages
          if num_messages < 0:
              raise Exception("Number of messages needs to be positive")
          msg_count = min(num_messages, len(mocked_messages))
          returned_messages = mocked_messages[:msg_count]
          mocked_messages = mocked_messages[msg_count:]
          return returned_messages
  
      mock_consumer.consume = mock_consume
      mock_get_consumer.return_value = mock_consumer
  
      operator = ConsumeFromTopicOperator(
          kafka_config_id="kafka_d",
          topics=["test"],
          task_id="test",
          poll_timeout=0.0001,
      )
  
      # execute the operator (this is essentially a no op as we're mocking the consumer)
      operator.execute(context={})
  
file_path: airflow/tests/providers/apache/kafka/operators/test_consume.py
imports: from __future__ import annotations
  import json
  import logging
  from typing import Any
  from unittest import mock
  import pytest
  from airflow.models import Connection
  from airflow.providers.apache.kafka.operators.consume import ConsumeFromTopicOperator
  from airflow.utils import db
label_desc: Use image node_id 4 for calling the TestConsumeFromTopic obj's underlying member method code with example usage: obj.test_operator_consume_max(mock_get_consumer) and returns: returned_messages