ZSTD support for faststream.

[jsripradha]

While using faststream consumer with zstd compression getting an error 76.
Added cramjam

[roma-frolov]

please show your MRE so that we can reproduce the error

[roma-frolov]

and please specify the os version, the faststream version, and the zstd library version.

[jsripradha]

@roma-frolov - Use MAC OS. Latest fastream and latest cramjam (as AIOkafka uses cramjam)
Error: FetchResponse_v4(throttle_time_ms=0, topics=[(topics='APM0018827-cr-global-search-DEV-indexing', partitions=[(partition=4, error_code=76, highwater_offset=-1, last_stable_offset=-1, aborted_transactions=[], message_set=None)])])

Issue is only during consumption. Able to produce message to topic
Kafka-python: confluentinc/confluent-kafka-python#858 (comment)
MRE (AIOkafka):
ssl_context = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
ssl_context.load_cert_chain(
certfile='client.pem',
keyfile='key.pem'
)
ssl_context.load_verify_locations('root.pem')
async def consume_messages():
# Create consumer
consumer = AIOKafkaConsumer(
bootstrap_servers=KAFKA_BROKERS,
group_id=GROUP_ID,
auto_offset_reset='earliest',
enable_auto_commit=True,
security_protocol="SSL",
ssl_context=ssl_context,
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
max_poll_records=500
)

try:
    consumer.subscribe([TOPIC])
    await consumer.start()
    # Consume messages
    async for message in consumer:
        print(f"Received message: "
              f"Topic={message.topic}, "
              f"Partition={message.partition}, "
              f"Offset={message.offset}, "
              f"Key={message.key}, "
              f"Value={message.value.decode('utf-8')}")

except KafkaError as e:
    print(f"Kafka error: {e}")
finally:
    # Stop the consumer
    await consumer.stop()

MRE (faststream):
broker = KafkaBroker(
bootstrap_servers=KAFKA_BROKERS,
security = security,
compression_type = "zstd",
client_id = "service"
)
@broker.subscriber("gtest")
async def process_message(message: str):
return {"response": "Hello, There!"}

app = FastStream(broker)

[jsripradha]

@Lancetnik

[Lancetnik]

@jsripradha did you check this project? https://github.com/ulbwa/faststream-compressors
It could be a reference for any compression integration

[jsripradha]

@Lancetnik this doesn't have zstd

[Lancetnik]

@Lancetnik this doesn't have zstd

It has a protocol, allows you to implement any compression middlewares you want