code: def test_kafka_producer_proper_record_metadata(
      kafka_broker, compression
  ):
      if compression == "zstd" and env_kafka_version() < (2, 1, 0):
          pytest.skip("zstd requires 2.1.0 or more")
      connect_str = ":".join(
          [kafka_broker.host, str(kafka_broker.port)]
      )
      producer = KafkaProducer(
          bootstrap_servers=connect_str,
          retries=5,
          max_block_ms=30000,
          compression_type=compression,
      )
      magic = producer._max_usable_produce_magic()
  
      # record headers are supported in 0.11.0
      if env_kafka_version() < (0, 11, 0):
          headers = None
      else:
          headers = [("Header Key", b"Header Value")]
  
      topic = random_string(5)
      future = producer.send(
          topic,
          value=b"Simple value",
          key=b"Simple key",
          headers=headers,
          timestamp_ms=9999999,
          partition=0,
      )
      record = future.get(timeout=5)
      assert record is not None
      assert record.topic == topic
      assert record.partition == 0
      assert record.topic_partition == TopicPartition(topic, 0)
      assert record.offset == 0
      if magic >= 1:
          assert record.timestamp == 9999999
      else:
          assert record.timestamp == -1  # NO_TIMESTAMP
  
      if magic >= 2:
          assert record.checksum is None
      elif magic == 1:
          assert record.checksum == 1370034956
      else:
          assert record.checksum == 3296137851
  
      assert record.serialized_key_size == 10
      assert record.serialized_value_size == 12
      if headers:
          assert record.serialized_header_size == 22
  
      if magic == 0:
          pytest.skip(
              "generated timestamp case is skipped for broker 0.9 and below"
          )
      send_time = time.time() * 1000
      future = producer.send(
          topic,
          value=b"Simple value",
          key=b"Simple key",
          timestamp_ms=None,
          partition=0,
      )
      record = future.get(timeout=5)
      assert abs(record.timestamp - send_time) <= 1000
  
file_path: kafka-python/test/test_producer.py
imports: import gc
  import platform
  import time
  import threading
  import pytest
  from kafka import KafkaConsumer, KafkaProducer, TopicPartition
  from kafka.producer.buffer import SimpleBufferPool
  from test.testutil import env_kafka_version, random_string
label_desc: Use image node_id 4 for calling a global function with example usage: test_kafka_producer_proper_record_metadata(kafka_broker, compression) without return types