Merge pull request #777 from Aiven-Open/matyaskuti/backups_headers

Do not skip None keyed headers on restore

karapace/backup/api.py

@@ -390,7 +390,7 @@ def _handle_producer_send(
             key=instruction.key,
             value=instruction.value,
             partition=instruction.partition_index,
-            headers=[(key.decode(), value) for key, value in instruction.headers if key is not None],
+            headers=[(key.decode() if key is not None else None, value) for key, value in instruction.headers],
             timestamp=instruction.timestamp,
         ).add_done_callback(producer_callback)
     except (KafkaError, AssertionError) as exc:

karapace/kafka/producer.py

@@ -32,7 +32,7 @@ class ProducerSendParams(TypedDict, total=False):
     key: str | bytes | None
     partition: int
     timestamp: int | None
-    headers: dict[str, bytes | None] | list[tuple[str, bytes | None]] | None
+    headers: dict[str | None, bytes | None] | list[tuple[str | None, bytes | None]] | None
 
 
 class KafkaProducer(_KafkaConfigMixin, Producer):

stubs/confluent_kafka/cimpl.pyi

@@ -51,7 +51,7 @@ class Producer:
         partition: int = -1,
         on_delivery: Callable[[KafkaError, Message], Any] | None = None,
         timestamp: int | None = -1,
-        headers: dict[str, bytes | None] | list[tuple[str, bytes | None]] | None = None,
+        headers: dict[str | None, bytes | None] | list[tuple[str | None, bytes | None]] | None = None,
     ) -> None: ...
     def flush(self, timeout: float = -1) -> None: ...
     def list_topics(self, topic: str | None = None, timeout: float = -1) -> ClusterMetadata: ...

tests/integration/kafka/test_producer.py

@@ -19,7 +19,7 @@ def test_send(self, producer: KafkaProducer, new_topic: NewTopic) -> None:
         value = b"value"
         partition = 0
         timestamp = int(time.time() * 1000)
-        headers = [(b"something", b"123")]
+        headers = [("something", b"123"), (None, "foobar")]
 
         fut = producer.send(
             new_topic.topic,