code: def test_end_to_end(kafka_broker, compression):
      if compression == "lz4":
          if env_kafka_version() < (0, 8, 2):
              pytest.skip("LZ4 requires 0.8.2")
          elif platform.python_implementation() == "PyPy":
              pytest.skip(
                  "python-lz4 crashes on older versions of pypy"
              )
  
      if compression == "zstd" and env_kafka_version() < (2, 1, 0):
          pytest.skip("zstd requires kafka 2.1.0 or newer")
  
      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,
          value_serializer=str.encode,
      )
      consumer = KafkaConsumer(
          bootstrap_servers=connect_str,
          group_id=None,
          consumer_timeout_ms=30000,
          auto_offset_reset="earliest",
          value_deserializer=bytes.decode,
      )
  
      topic = random_string(5)
  
      messages = 100
      futures = []
      for i in range(messages):
          futures.append(producer.send(topic, "msg %d" % i))
      ret = [f.get(timeout=30) for f in futures]
      assert len(ret) == messages
      producer.close()
  
      consumer.subscribe([topic])
      msgs = set()
      for i in range(messages):
          try:
              msgs.add(next(consumer).value)
          except StopIteration:
              break
  
      assert msgs == set(["msg %d" % (i,) for i in range(messages)])
      consumer.close()
  
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 2 for calling a global function with example usage: test_end_to_end(kafka_broker, compression) without return types