chore: add configurable pagination key, update readme and recipe (#3)

Author: cgpoh

README.md

@@ -1,5 +1,8 @@
 # datahub-apicurio-kafka
 ApiCurio Schema Registry implementation for [DataHub Kafka ingestion source](https://datahubproject.io/docs/generated/ingestion/sources/kafka/). Only support AVRO schema currently.
 
+## Sample Ingestion Recipe
+A sample recipe can be found [here](./kafka_src_recipe.yaml)
+
 ## Installation
 Package can be installed using `pip install apicurio-datahub-kafka`

kafka_src_recipe.yaml

@@ -6,5 +6,7 @@ source:
     connection:
       bootstrap: "localhost:9093"
       schema_registry_url: http://localhost:8085/apis/registry/v2
+      schema_registry_config:
+        pagination: 100
 #sink:
 #  type: "console"

setup.py

@@ -2,7 +2,7 @@
 
 setup_output = setup(
     name="apicurio_datahub_kafka",
-    version="1.1.2",
+    version="1.1.5",
     description="DataHub ApiCurio Schema Registry for Kafka Source",
     package_dir={"": "src"},
     packages=find_packages("src"),

src/apicurio_kafka/apicurio_schema_registry.py

@@ -3,6 +3,8 @@
 import logging
 
 import nest_asyncio
+from apicurioregistrysdk.client.search.artifacts.artifacts_request_builder import ArtifactsRequestBuilder
+
 nest_asyncio.apply()
 
 from hashlib import md5
@@ -46,8 +48,26 @@ async def _async_init(
         self.registry_client = RegistryClient(request_adapter)
 
         try:
-            self.known_schema_registry_subjects: ArtifactSearchResults = await self.registry_client.search.artifacts.get()
+            limit = self.source_config.connection.schema_registry_config.get("pagination")
+            if limit == None:
+                logger.warning(f"Unable to get pagination key in recipe. Setting maximum number of artifacts to be returned to default 20")
+                limit = 20
+            else:
+                logger.info(f"Maximum number of artifacts to be returned: {limit}")
+
+            query_params = ArtifactsRequestBuilder.ArtifactsRequestBuilderPostQueryParameters(
+                limit=limit
+            )
+
+            request_configuration = (
+                ArtifactsRequestBuilder.ArtifactsRequestBuilderPostRequestConfiguration(
+                    query_parameters=query_params
+                )
+            )
+            self.known_schema_registry_subjects: ArtifactSearchResults = await self.registry_client.search.artifacts.get(request_configuration=request_configuration)
             logger.info(f"Known schema registry subjects: {self.known_schema_registry_subjects.count}")
+            for artifact in self.known_schema_registry_subjects.artifacts:
+                logger.info(f"known artifact id: {artifact.id}")
         except Exception as e:
             logger.warning(f"Failed to get artifacts from schema registry: {e}")
 
@@ -109,15 +129,15 @@ async def _get_schema_and_fields(
         # Obtain the schema fields from schema for the topic.
         fields: List[SchemaField] = []
         if artifact is not None:
-            logger.debug(
+            logger.info(
                 f"The {schema_type_str} schema subject:'{artifact.id}', grp ID:'{artifact.group_id}' is found for topic:'{topic}'."
             )
             fields = await self._get_schema_fields(
                 topic=topic, artifact=artifact, is_key_schema=is_key_schema
             )
         else:
-            logger.debug(
-                f"For topic: {topic}, the schema registry subject for the schema is not found."
+            logger.info(
+                f"For topic: {topic}, the {schema_type_str} schema registry subject for the schema is not found."
             )
         return artifact, fields
 

tests/apicurio/test_schema_registry.py

@@ -91,6 +91,9 @@ async def test_apicurio_schema_registry():
 
     kafka_source_config = KafkaSourceConfig()
     kafka_source_config.connection.schema_registry_url = f"http://{os.environ[REGISTRY_HOST]}:{os.environ[REGISTRY_PORT]}/apis/registry/v2"
+    kafka_source_config.connection.schema_registry_config = {
+        "pagination": 100
+    }
 
     kafka_source_report = KafkaSourceReport()