code: def test__send_offset_requests(fetcher, mocker):
      tp = TopicPartition("topic_send_offset", 1)
      mocked_send = mocker.patch.object(fetcher, "_send_offset_request")
      send_futures = []
  
      def send_side_effect(*args, **kw):
          f = Future()
          send_futures.append(f)
          return f
  
      mocked_send.side_effect = send_side_effect
  
      mocked_leader = mocker.patch.object(
          fetcher._client.cluster, "leader_for_partition"
      )
      # First we report unavailable leader 2 times different ways and later
      # always as available
      mocked_leader.side_effect = itertools.chain(
          [None, -1], itertools.cycle([0])
      )
  
      # Leader == None
      fut = fetcher._send_offset_requests({tp: 0})
      assert fut.failed()
      assert isinstance(fut.exception, StaleMetadata)
      assert not mocked_send.called
  
      # Leader == -1
      fut = fetcher._send_offset_requests({tp: 0})
      assert fut.failed()
      assert isinstance(fut.exception, LeaderNotAvailableError)
      assert not mocked_send.called
  
      # Leader == 0, send failed
      fut = fetcher._send_offset_requests({tp: 0})
      assert not fut.is_done
      assert mocked_send.called
      # Check that we bound the futures correctly to chain failure
      send_futures.pop().failure(NotLeaderForPartitionError(tp))
      assert fut.failed()
      assert isinstance(fut.exception, NotLeaderForPartitionError)
  
      # Leader == 0, send success
      fut = fetcher._send_offset_requests({tp: 0})
      assert not fut.is_done
      assert mocked_send.called
      # Check that we bound the futures correctly to chain success
      send_futures.pop().success({tp: (10, 10000)})
      assert fut.succeeded()
      assert fut.value == {tp: (10, 10000)}
  
imports: from __future__ import absolute_import
  import pytest
  from collections import OrderedDict
  import itertools
  import time
  from kafka.client_async import KafkaClient
  from kafka.consumer.fetcher import CompletedFetch, ConsumerRecord, Fetcher, NoOffsetForPartitionError
  from kafka.consumer.subscription_state import SubscriptionState
  from kafka.future import Future
  from kafka.metrics import Metrics
  from kafka.protocol.fetch import FetchRequest, FetchResponse
  from kafka.protocol.offset import OffsetResponse
  from kafka.errors import StaleMetadata, LeaderNotAvailableError, NotLeaderForPartitionError, UnknownTopicOrPartitionError, OffsetOutOfRangeError
  from kafka.record.memory_records import MemoryRecordsBuilder, MemoryRecords
  from kafka.structs import OffsetAndMetadata, TopicPartition
label_desc: Use image node_id 10 for calling a global function with example usage: test__send_offset_requests(fetcher, mocker) and returns: f