Merge pull request #484 from aiven/anatolii/backup-restore-fix

Handling None properly during backup

Author: jjaakola-aiven

karapace/schema_backup.py

@@ -140,7 +140,6 @@ def _create_schema_topic_if_needed(self) -> None:
             return
 
         self.init_admin_client()
-
         start_time = time.monotonic()
         wait_time = constants.MINUTE
         while True:
@@ -210,9 +209,10 @@ def _restore_backup_version_1_single_array(self, fp: IO) -> None:
 
     def _restore_backup_version_2(self, fp: IO) -> None:
         for line in fp:
-            hex_key, hex_value = line.split("\t")
-            key = base64.b16decode(hex_key).decode("utf8")
-            value = base64.b16decode(hex_value.strip()).decode("utf8")  # strip to remove the linefeed
+            hex_key, hex_value = [val.strip() for val in line.split("\t")]  # strip to remove the linefeed
+
+            key = base64.b16decode(hex_key).decode("utf8") if hex_key != "null" else hex_key
+            value = base64.b16decode(hex_value.strip()).decode("utf8") if hex_value != "null" else hex_value
             self._handle_restore_message((key, value))
 
     def export(self, export_func) -> None:
@@ -240,9 +240,9 @@ def export(self, export_func) -> None:
             LOG.info("Schema export written to stdout")
         self.close()
 
-    def encode_key(self, key: Optional[Union[JsonData, str]]) -> bytes:
-        if not key:
-            return b""
+    def encode_key(self, key: Optional[Union[JsonData, str]]) -> Optional[bytes]:
+        if key == "null":
+            return None
         if not self.key_formatter:
             if isinstance(key, str):
                 return key.encode("utf8")
@@ -261,12 +261,8 @@ def encode_value(value: Union[JsonData, str]) -> Optional[bytes]:
 
 
 def serialize_record(key_bytes: Optional[bytes], value_bytes: Optional[bytes]) -> str:
-    key = b""
-    if key_bytes is not None:
-        key = base64.b16encode(key_bytes).decode("utf8")
-    value = b""
-    if value_bytes is not None:
-        value = base64.b16encode(value_bytes).decode("utf8")
+    key = base64.b16encode(key_bytes).decode("utf8") if key_bytes is not None else "null"
+    value = base64.b16encode(value_bytes).decode("utf8") if value_bytes is not None else "null"
     return f"{key}\t{value}\n"
 
 

tests/integration/test_data/test_restore_non_schema_topic_v2.log

@@ -3,3 +3,5 @@
 6B65795F74776F	76616C75655F74776F
 	76616C75655F7468726565
 6B65795F666F7572	
+6B65795F666F7572	null
+null	76616C75655F7468726565

tests/integration/test_schema_backup.py

@@ -7,6 +7,7 @@
 from kafka import KafkaConsumer
 from karapace.client import Client
 from karapace.config import set_config_defaults
+from karapace.kafka_rest_apis import KafkaRestAdminClient
 from karapace.key_format import is_key_in_canonical_format
 from karapace.schema_backup import SchemaBackup, serialize_record
 from karapace.utils import Expiration
@@ -69,27 +70,26 @@ async def test_backup_get(
 
 
 async def test_backup_restore_and_get_non_schema_topic(
-    kafka_servers: KafkaServers,
-    tmp_path: Path,
+    kafka_servers: KafkaServers, tmp_path: Path, admin_client: KafkaRestAdminClient
 ) -> None:
     test_topic_name = new_random_name("non-schemas")
 
     config = set_config_defaults(
         {
             "bootstrap_uri": kafka_servers.bootstrap_servers,
-            "topic_name": test_topic_name,
         }
     )
+    admin_client.new_topic(name=test_topic_name)
 
     # Restore from backup
     test_data_path = Path("tests/integration/test_data/")
     restore_location = test_data_path / "test_restore_non_schema_topic_v2.log"
-    sb = SchemaBackup(config, str(restore_location))
+    sb = SchemaBackup(config, str(restore_location), topic_option=test_topic_name)
     sb.restore_backup()
 
     # Get the backup
     backup_location = tmp_path / "non_schemas_topic.log"
-    sb = SchemaBackup(config, str(backup_location))
+    sb = SchemaBackup(config, str(backup_location), topic_option=test_topic_name)
     sb.export(serialize_record)
     # The backup file has been created
     assert os.path.exists(backup_location)