code: def test__send_offset_requests_multiple_nodes(fetcher, mocker):
      tp1 = TopicPartition("topic_send_offset", 1)
      tp2 = TopicPartition("topic_send_offset", 2)
      tp3 = TopicPartition("topic_send_offset", 3)
      tp4 = TopicPartition("topic_send_offset", 4)
      mocked_send = mocker.patch.object(fetcher, "_send_offset_request")
      send_futures = []
  
      def send_side_effect(node_id, timestamps):
          f = Future()
          send_futures.append((node_id, timestamps, f))
          return f
  
      mocked_send.side_effect = send_side_effect
  
      mocked_leader = mocker.patch.object(
          fetcher._client.cluster, "leader_for_partition"
      )
      mocked_leader.side_effect = itertools.cycle([0, 1])
  
      # -- All node succeeded case
      tss = OrderedDict([(tp1, 0), (tp2, 0), (tp3, 0), (tp4, 0)])
      fut = fetcher._send_offset_requests(tss)
      assert not fut.is_done
      assert mocked_send.call_count == 2
  
      req_by_node = {}
      second_future = None
      for node, timestamps, f in send_futures:
          req_by_node[node] = timestamps
          if node == 0:
              # Say tp3 does not have any messages so it's missing
              f.success({tp1: (11, 1001)})
          else:
              second_future = f
      assert req_by_node == {0: {tp1: 0, tp3: 0}, 1: {tp2: 0, tp4: 0}}
  
      # We only resolved 1 future so far, so result future is not yet ready
      assert not fut.is_done
      second_future.success({tp2: (12, 1002), tp4: (14, 1004)})
      assert fut.succeeded()
      assert fut.value == {
          tp1: (11, 1001),
          tp2: (12, 1002),
          tp4: (14, 1004),
      }
  
      # -- First succeeded second not
      del send_futures[:]
      fut = fetcher._send_offset_requests(tss)
      assert len(send_futures) == 2
      send_futures[0][2].success({tp1: (11, 1001)})
      send_futures[1][2].failure(UnknownTopicOrPartitionError(tp1))
      assert fut.failed()
      assert isinstance(fut.exception, UnknownTopicOrPartitionError)
  
      # -- First fails second succeeded
      del send_futures[:]
      fut = fetcher._send_offset_requests(tss)
      assert len(send_futures) == 2
      send_futures[0][2].failure(UnknownTopicOrPartitionError(tp1))
      send_futures[1][2].success({tp1: (11, 1001)})
      assert fut.failed()
      assert isinstance(fut.exception, UnknownTopicOrPartitionError)
  
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 11 for calling a global function with example usage: test__send_offset_requests_multiple_nodes(fetcher, mocker) and returns: f