Repository: lsst-sqre/strimzi-registry-operator

Issue:
Update and create policy for updating the Confluent Schema Registry image
We currently hardcode the Schema Registry image version:

https://github.com/lsst-sqre/strimzi-registry-operator/blob/72a2be9919664ba3cb9adf2afa44c987e0e9bd56/strimziregistryoperator/deployments.py#L167

We should develop a pattern for updating this/maintaining this:

- In new versions of strimzi-registry-operator, update the default image.
- In StrimziSchemaRegistry, add a "registryImage" configuration field that allows a user to independently set the image version, replacing the default.

Discussion:

Patch:
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..9cd54e4
--- /dev/null
+++ b/README.md
@@ -0,0 +1,296 @@
+# strimzi-registry-operator
+
+A Kubernetes Operator for running the [Confluent Schema Registry](https://docs.confluent.io/current/schema-registry/index.html) in a [Strimzi](https://strimzi.io)-based [Kafka](https://kafka.apache.org/) cluster that's optionally secured with TLS.
+
+Overview:
+
+- Once you deploy a `StrimziSchemaRegistry` resource, the operator creates a Kubernetes deployment of the Confluent Schema Registry, along with an associated Kubernetes service and secret.
+- Works with Strimzi's TLS authentication and authorization by converting the TLS certificate associated with a KafkaUser into a JKS-formatted keystore and truststore that's used by Confluence Schema Registry.
+- When Strimzi updates either the Kafka cluster's CA certificate or the KafkaUser's client certificates, the operator automatically recreates the JKS truststore/keystore secrets and triggers a rolling restart of the Schema Registry pods.
+
+## Deploy the operator
+
+There are two steps for running strimzi-registry-operator.
+First, you'll need to deploy the operator itself — this is described in this section.
+Once the operator is deployed, you can deploy `StrimziSchemaRegistry` resources that actually create and maintain a Confluent Schema Registry application (this is discussed in later sections).
+
+Two operator deployment options are available: [Helm](https://helm.sh) and [Kustomize](https://kustomize.io).
+
+### With Helm
+
+A Helm chart is available for strimzi-registry-operator on GitHub at [lsst-sqre/charts](https://github.com/lsst-sqre/charts/tree/master/charts/strimzi-registry-operator):
+
+```sh
+helm repo add lsstsqre https://lsst-sqre.github.io/charts/
+helm repo update
+helm install lsstsqre/strimzi-registry-operator --name ssr --set watchNamespace="...",clusterName="..."
+```
+
+[See the Helm chart's README](https://github.com/lsst-sqre/charts/tree/master/charts/strimzi-registry-operator) for important values to set, including the names of the Strimzi Kafka cluster and namespace for `KafkaUser` resources to watch.
+
+### With Kustomize
+
+The manifests for the operator itself are located in the `/manifests` directory of this repository.
+You can use Kustomize to build a single YAML file for deployment:
+
+```sh
+kustomize build manifests > manifest.yaml
+kubectl apply -f manifest.yaml
+```
+
+To configure the name of the Strimzi cluster and the namespace where `KafkaUser` resources are available, you'll need to create your own Kustomize overlay.
+
+A basic ``kustomization.yaml`` file is:
+
+```yaml
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  - github.com/lsst-sqre/strimzi-registry-operator.git//manifests?ref=0.4.1
+
+patches:
+  - strimzi-registry-operator-deployment.yaml
+```
+
+Use the `strimzi-registry-operator-deployment.yaml` patch to set environment variables:
+
+```yaml
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: strimzi-registry-operator
+spec:
+  template:
+    spec:
+      containers:
+        - name: operator
+          env:
+            - name: SSR_CLUSTER_NAME
+              value: events
+            - name: SSR_NAMESPACE
+              value: events
+```
+
+- `SSR_CLUSTER_NAME` is the name of the Strimzi Kafka cluster.
+- `SSR_NAMESPACE` is the namespace where the Strimzi Kafka cluster is deployed and where `KafkaUser` resources are found.
+
+## Deploy a Schema Registry
+
+### Step 1. Deploy a KafkaTopic
+
+Deploy a `KafkaTopic` that the Schema Registry will use as its primary storage.
+
+```yaml
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaTopic
+metadata:
+  name: registry-schemas
+  labels:
+    strimzi.io/cluster: events
+spec:
+  partitions: 1
+  replicas: 3
+  config:
+    # http://kafka.apache.org/documentation/#topicconfigs
+    cleanup.policy: compact
+```
+
+> **Note**
+> The name `registry-schemas` is currently required.
+> The default name, `_schemas` isn't used because it isn't convenient to create with `KafkaTopic` resources.
+
+### Step 2. Deploy a KafkaUser
+
+Deploy a `KafkaUser` for the Schema Registry that gives the Schema Registry sufficient permissions:
+
+```yaml
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaUser
+metadata:
+  name: confluent-schema-registry
+  labels:
+    strimzi.io/cluster: events
+spec:
+  authentication:
+    type: tls
+  authorization:
+    # Official docs on authorizations required for the Schema Registry:
+    # https://docs.confluent.io/current/schema-registry/security/index.html#authorizing-access-to-the-schemas-topic
+    type: simple
+    acls:
+      # Allow all operations on the registry-schemas topic
+      # Read, Write, and DescribeConfigs are known to be required
+      - resource:
+          type: topic
+          name: registry-schemas
+          patternType: literal
+        operation: All
+        type: allow
+      # Allow all operations on the schema-registry* group
+      - resource:
+          type: group
+          name: schema-registry
+          patternType: prefix
+        operation: All
+        type: allow
+      # Allow Describe on the __consumer_offsets topic
+      - resource:
+          type: topic
+          name: __consumer_offsets
+          patternType: literal
+        operation: Describe
+        type: allow
+```
+
+### Step 3. Deploy the StrimziSchemaRegistry
+
+Now that there is a topic and a user, you can deploy the Schema Registry itself.
+The strimzi-schema-registry operator deploys the Schema Registry given a `StrimziSchemaRegistry` resource:
+
+```yaml
+apiVersion: roundtable.lsst.codes/v1beta1
+kind: StrimziSchemaRegistry
+metadata:
+  name: confluent-schema-registry
+spec:
+  strimzi-version: v1beta2
+  listener: tls
+```
+
+The next section describes the configuration properties for the `StrimziSchemaRegistry`.
+
+## StrimziSchemaRegistry configuration properties
+
+- `strimziVersion` is the version of the `kafka.strimzi.io` Custom Resource API to use.
+  The correct value depends on the deployed version of Strimzi.
+  The current Strimzi API  version is `v1beta2`.
+  Strimzi versions 0.21.0 and earlier support the `v1beta1` API.
+  (A deprecated version of the configuration is `strimzi-version`.)
+
+- `listener` is the **name** of the Kafka listener that the Schema Registry should use.
+  The default value is `tls`, but you should set this value based on your `Kafka` resource.
+
+- `registryImage` is the name of the Confluent Schema Registry Docker image (without the tag).
+  Default is `confluentinc/cp-schema-registry`.
+
+- `registryImageTag` is the name of the Schema Registry Docker image's tag.
+  Use this property to change the version of the Confluent Schema Registry that you're deploying through the `StrimziSchemaRegistry`.
+  Default is `7.2.1`.
+
+- `cpuLimit` is the cap on CPU usage for the Schema Registry container. Default is to leave unset. Example `1000m` limits to 1 CPU.
+
+- `cpuRequest` is the requested CPU for the Schema Registry container. Default is to leave unset. Example: `100m` requests 0.1 CPU.
+
+- `memoryLimit` is the cap on memory usage for the Schema Registry container. Default is to leave unset. Example `1000M` limits to 1000 megabytes.
+
+- `memoryRequest` is the requested memory for the Schema Registry container. Default is to leave unset. Example: `768M` requests 768 megabytes.
+
+- `compatibilityLevel` is the default schema compatibility level, corresponding to [schema.compatibility.level](https://docs.confluent.io/platform/current/schema-registry/installation/config.html#schema-compatibility-level). Default is "forward". Possible values:
+
+  - `none`
+  - `backward`
+  - `backward_transitive`
+  - `forward`
+  - `forward_transitive`
+  - `full`
+  - `full_transitive`
+
+- `securityProtocol` is the security protocol for the Schema Registry to communicate with Kafka. Default is SSL. Can be:
+  
+  - `SSL`
+  - `PLAINTEXT`
+  - `SASL_PLAINTEXT`
+  - `SASL_SSL`
+
+### The listener configuration
+
+The `spec.listener` field in the `StrimziSchemaRegistry` resource specifies the Kafka broker listener that the Schema Registry uses.
+These listeners are configured in the `Kafka` resource you created with Strimzi.
+
+Consider a `Kafka` resource:
+
+```yaml
+apiVersion: kafka.strimzi.io/v1beta2
+kind: Kafka
+metadata:
+  name: my-cluster
+spec:
+  kafka:
+    #...
+    listeners:
+      - name: plain
+        port: 9092
+        type: internal
+        tls: false
+      - name: tls
+        port: 9093
+        type: internal
+        tls: true
+        authentication:
+          type: tls
+      - name: external1
+        port: 9094
+        type: route
+        tls: true
+      - name: external2
+        port: 9095
+        type: ingress
+        tls: true
+        authentication:
+          type: tls
+        configuration:
+          bootstrap:
+            host: bootstrap.myingress.com
+          brokers:
+          - broker: 0
+            host: broker-0.myingress.com
+          - broker: 1
+            host: broker-1.myingress.com
+          - broker: 2
+            host: broker-2.myingress.com
+    #...
+```
+
+To use the encrypted internal listener, the `spec.listener` field in your `StrimziSchemaRegistry` resource should be `tls`:
+
+```yaml
+apiVersion: roundtable.lsst.codes/v1beta1
+kind: StrimziSchemaRegistry
+metadata:
+  name: confluent-schema-registry
+spec:
+  listener: tls
+```
+
+To use the unencrypted internal listener instead, the `spec.listener` field in your `StrimziSchemaRegistry` resource should be `plain` instead:
+
+```yaml
+apiVersion: roundtable.lsst.codes/v1beta1
+kind: StrimziSchemaRegistry
+metadata:
+  name: confluent-schema-registry
+spec:
+  listener: plain
+```
+
+### Strimzi `v1beta1` listener configuration
+
+In older versions of Strimzi with the `v1beta1` API, listeners were not named.
+Instead, three types of listeners were available:
+
+```yaml
+apiVersion: kafka.strimzi.io/v1beta1
+kind: Kafka
+spec:
+  kafka:
+    # ...
+    listeners:
+      plain: {}
+      tls:
+        authentication:
+          type: "tls"
+      external: {}
+```
+
+In this case, set the `spec.listener` field in your `StrimziSchemaRegistry` to either `plain`, `tls`, or `external`.
diff --git a/README.rst b/README.rst
deleted file mode 100644
index d9dda56..0000000
--- a/README.rst
+++ /dev/null
@@ -1,271 +0,0 @@
-#########################
-strimzi-registry-operator
-#########################
-
-A Kubernetes Operator for running the `Confluent Schema Registry <https://docs.confluent.io/current/schema-registry/index.html>`_ in a `Strimzi <https://strimzi.io>`_-based `Kafka <https://kafka.apache.org/>`_ cluster that's optionally secured with TLS.
-
-Overview:
-
-- Once you deploy a ``StrimziSchemaRegistry`` manifest, the operator creates a Kubernetes deployment of the Confluent Schema Registry, along with an associated Kubernetes service and secret.
-- Works with Strimzi's TLS authentication and authorization by converting the TLS certificate associated with a KafkaUser into a JKS-formatted keystore and truststore that's used by Confluence Schema Registry.
-- When Strimzi updates either the Kafka cluster's CA certificate or the KafkaUser's client certificates, the operator automatically recreates the JKS truststore/keystore secrets and triggers a rolling restart of the Schema Registry pods.
-
-Deploy the operator
-===================
-
-There are two steps for running strimzi-registry-operator. First, you'll need to deploy the operator itself --- this is described in this section.
-Once the operator is deployed, you can deploy ``StrimziSchemaRegistry`` resources that actually create and maintain a Confluent Schema Registry application (this is discussed in later sections).
-
-Two operator deployment options are available: `Helm <https://helm.sh>`__ and `Kustomize <https://kustomize.io>`__.
-
-With Helm
----------
-
-A Helm chart is available for strimzi-registry-operator on GitHub at `lsst-sqre/charts <https://github.com/lsst-sqre/charts/tree/master/charts/strimzi-registry-operator>`_.
-
-.. code-block:: sh
-
-   helm repo add lsstsqre https://lsst-sqre.github.io/charts/
-   helm repo update
-   helm install lsstsqre/strimzi-registry-operator --name ssr --set watchNamespace="...",clusterName="..."
-
-`See the Helm chart's README <lsst-sqre/charts <https://github.com/lsst-sqre/charts/tree/master/charts/strimzi-registry-operator>`__ for important values to set, including the names of the Strimzi Kafka cluster and namespace for KafkaUser resources to watch.
-
-With Kustomize
---------------
-
-The manifests for the operator itself are located in the `/manifests` directory of this repository.
-You can use Kustomize to build a single YAML file for deployment.::
-
-    kustomize build manifests > manifest.yaml
-    kubectl apply -f manifest.yaml
-
-To configure the name of the Strimzi cluster and the namespace where KafkaUser resources are available, you'll need to create your own Kustomize overlay.
-
-A basic ``kustomization.yaml`` file is:
-
-.. code-block:: yaml
-
-   apiVersion: kustomize.config.k8s.io/v1beta1
-   kind: Kustomization
-
-   resources:
-     - github.com/lsst-sqre/strimzi-registry-operator.git//manifests?ref=0.4.1
-
-   patches:
-     - strimzi-registry-operator-deployment.yaml
-
-Use the ``strimzi-registry-operator-deployment.yaml`` patch to set environment variables:
-
-.. code-block:: yaml
-
-   apiVersion: apps/v1
-   kind: Deployment
-   metadata:
-     name: strimzi-registry-operator
-   spec:
-     template:
-       spec:
-         containers:
-           - name: operator
-             env:
-               - name: SSR_CLUSTER_NAME
-                 value: events
-               - name: SSR_NAMESPACE
-                 value: events
-
-- ``SSR_CLUSTER_NAME`` is the name of the Strimzi Kafka cluster.
-- ``SSR_NAMESPACE`` is the namespace where the Strimzi Kafka cluster is deployed and where KafkaUser resources are found.
-
-Deploy a Schema Registry
-========================
-
-Step 1. Deploy a KafkaTopic
----------------------------
-
-Deploy a ``KafkaTopic`` that the Schema Registry will use as its primary storage.
-
-.. code-block:: yaml
-
-   apiVersion: kafka.strimzi.io/v1beta2
-   kind: KafkaTopic
-   metadata:
-     name: registry-schemas
-     labels:
-       strimzi.io/cluster: events
-   spec:
-     partitions: 1
-     replicas: 3
-     config:
-       # http://kafka.apache.org/documentation/#topicconfigs
-       cleanup.policy: compact
-
-.. important::
-
-   The name ``registry-schemas`` is currently required.
-   The default name, ``_schemas`` isn't used because it isn't convenient to create with ``KafkaTopic`` resources.
-
-Step 2. Deploy a KafkaUser
---------------------------
-
-Deploy a KafkaUser for the Schema Registry that gives the Schema Registry sufficient permissions:
-
-.. code-block:: yaml
-
-   apiVersion: kafka.strimzi.io/v1beta2
-   kind: KafkaUser
-   metadata:
-     name: confluent-schema-registry
-     labels:
-       strimzi.io/cluster: events
-   spec:
-     authentication:
-       type: tls
-     authorization:
-       # Official docs on authorizations required for the Schema Registry:
-       # https://docs.confluent.io/current/schema-registry/security/index.html#authorizing-access-to-the-schemas-topic
-       type: simple
-       acls:
-         # Allow all operations on the _schemas topic
-         # Read, Write, and DescribeConfigs are known to be required
-         - resource:
-             type: topic
-             name: registry-schemas
-             patternType: literal
-           operation: All
-           type: allow
-         # Allow all operations on the schema-registry* group
-         - resource:
-             type: group
-             name: schema-registry
-             patternType: prefix
-           operation: All
-           type: allow
-         # Allow Describe on the __consumer_offsets topic
-         # (The official docs also mention DescribeConfigs?)
-         - resource:
-             type: topic
-             name: __consumer_offsets
-             patternType: literal
-           operation: Describe
-           type: allow
-
-Step 3. Deploy the StrimziSchemaRegistry
-----------------------------------------
-
-Now that there is a topic and a user, you can deploy the Schema Registry itself.
-The strimzi-schema-registry operator deploys the Schema Registry given a ``StrimziSchemaRegistry`` resource:
-
-.. code-block:: yaml
-
-   apiVersion: roundtable.lsst.codes/v1beta1
-   kind: StrimziSchemaRegistry
-   metadata:
-     name: confluent-schema-registry
-   spec:
-     strimzi-version: v1beta2
-     listener: tls
-
-- ``strimzi-version`` is the version of the ``kafka.strimzi.io`` Custom Resource API to use.
-  The correct value depends on the deployed version of Strimzi.
-  The current Strimzi API  version is ``v1beta2``.
-  Strimzi versions 0.21.0 and earlier support the ``v1beta1`` API.
-
-- ``listener`` is the :ref:`name <listener-config>` of the Kafka listener that the Schema Registry should use.
-  The default value is ``tls``, but you should set this value based on your ``Kafka`` resource.
-
-.. _listener-config:
-
-The listener configuration
-""""""""""""""""""""""""""
-
-The ``spec.listener`` field in the ``StrimziSchemaRegistry`` resource specifies the Kafka broker listener that the Schema Registry uses.
-These listeners are configured in the ``Kafka`` resource you created with Strimzi.
-
-For example, consider a ``Kafka`` resource:
-
-.. code-block:: yaml
-
-   apiVersion: kafka.strimzi.io/v1beta2
-   kind: Kafka
-   metadata:
-     name: my-cluster
-   spec:
-     kafka:
-       #...
-       listeners:
-         - name: plain
-           port: 9092
-           type: internal
-           tls: false
-         - name: tls
-           port: 9093
-           type: internal
-           tls: true
-           authentication:
-             type: tls
-         - name: external1
-           port: 9094
-           type: route
-           tls: true
-         - name: external2
-           port: 9095
-           type: ingress
-           tls: true
-           authentication:
-             type: tls
-           configuration:
-             bootstrap:
-               host: bootstrap.myingress.com
-             brokers:
-             - broker: 0
-               host: broker-0.myingress.com
-             - broker: 1
-               host: broker-1.myingress.com
-             - broker: 2
-               host: broker-2.myingress.com
-       #...
-
-To use the encrypted internal listener, the ``spec.listener`` field in your ``StrimziSchemaRegistry`` resource should be ``tls``:
-
-.. code-block:: yaml
-
-   apiVersion: roundtable.lsst.codes/v1beta1
-   kind: StrimziSchemaRegistry
-   metadata:
-     name: confluent-schema-registry
-   spec:
-     listener: tls
-
-To use the unencrypted internal listener instead, the ``spec.listener`` field in your ``StrimziSchemaRegistry`` resource should be ``plain`` instead:
-
-.. code-block:: yaml
-
-   apiVersion: roundtable.lsst.codes/v1beta1
-   kind: StrimziSchemaRegistry
-   metadata:
-     name: confluent-schema-registry
-   spec:
-     listener: plain
-
-Strimzi ``v1beta1`` listener configuration
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-In older versions of Strimzi with the ``v1beta1`` API, listeners were not named.
-Instead, three types of listeners were available:
-
-.. code-block:: yaml
-
-   apiVersion: kafka.strimzi.io/v1beta1
-   kind: Kafka
-   spec:
-     kafka:
-       # ...
-       listeners:
-         plain: {}
-         tls:
-           authentication:
-             type: "tls"
-         external: {}
-
-In this case, set the ``spec.listener`` field in your ``StrimziSchemaRegistry`` to either ``plain``, ``tls``, or ``external``.
diff --git a/manifests/registry-crd.yaml b/manifests/registry-crd.yaml
index 1939999..e9a8247 100644
--- a/manifests/registry-crd.yaml
+++ b/manifests/registry-crd.yaml
@@ -21,12 +21,17 @@ spec:
               description: >-
                 The specification of the Schema Registry instance.
               properties:
-                strimzi-version:
+                strimziVersion:
                   type: string
                   default: "v1beta2"
                   description: >-
                     The version of the Strimzi Custom Resource API to use. The
                     correct value depends on the deployed version of Strimzi.
+                strimzi-version:
+                  type: string
+                  default: "v1beta2"
+                  description: >-
+                    Deprecated. Use strimziVersion instead.
                 listener:
                   type: string
                   default: "tls"
@@ -39,6 +44,59 @@ spec:
                     The type of service to create for the registry. Default is
                     ClusterIP. Can be NodePort to publish the registry
                     externally.
+                registryImage:
+                  type: string
+                  default: "confluentinc/cp-schema-registry"
+                  description: >-
+                    The Docker image for the Confluent Schema Registry.
+                registryImageTag:
+                  type: string
+                  default: "7.2.1"
+                  description: >-
+                    The Docker image tag for the Confluent Schema Registry.
+                    Use this property to control the version of the deployed
+                    schema registry.
+                cpuLimit:
+                  type: string
+                  default: ""
+                  description: >-
+                    The cap on CPU usage for the Schema Registry container.
+                    Default is to leave unset. Example 1000m limits to 1 CPU.
+                cpuRequest:
+                  type: string
+                  default: ""
+                  description: >-
+                    The requested CPU allocation for the Schema Registry
+                    container. Default is to leave unset. Example 1000m
+                    requests 1 CPU.
+                memoryLimit:
+                  type: string
+                  default: ""
+                  description: >-
+                    The cap on memory usage for the Schema Registry container.
+                    Default is to leave unset. Example 1000M limits to
+                    1000 megabytes.
+                memoryRequest:
+                  type: string
+                  default: ""
+                  description: >-
+                    The requested memory for the Schema Registry container.
+                    Default is to leave unset. Example: `768M` requests
+                    768 megabytes.
+                compatibilityLevel:
+                  type: string
+                  default: "forward"
+                  description: >-
+                    The default schema compatibility level in a subject.
+                    Can be none, backward, backward_transitive, forward,
+                    forward_transitive, full, and full_transitive.
+                securityProtocol:
+                  type: string
+                  default: "SSL"
+                  description: >-
+                    The security protocol for the Schema Registry to
+                    communicate with Kafka. Default is SSL. Can be SSL,
+                    PLAINTEXT, SASL_PLAINTEXT, or SASL_SSL.
   names:
     kind: StrimziSchemaRegistry
     plural: strimzischemaregistries
diff --git a/setup.cfg b/setup.cfg
index 4a03555..6906572 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -3,8 +3,8 @@ name = strimzi-registry-operator
 description = A Kubernetes Operator for deploying Confluent Schema Registry servers into a Strimzi-based Kafka cluster.
 author = Association of Universities for Research in Astronomy, Inc. (AURA)
 author_email = sqre-admin@lists.lsst.org
-long_description = file: README.rst, LICENSE
-long_description_content_type = text/x-rst
+long_description = file: README.md, LICENSE
+long_description_content_type = text/markdown
 license = MIT
 url = https://github.com/lsst-sqre/strimzi-registry-operator
 project_urls =
diff --git a/strimziregistryoperator/deployments.py b/strimziregistryoperator/deployments.py
index da8f755..b943a18 100644
--- a/strimziregistryoperator/deployments.py
+++ b/strimziregistryoperator/deployments.py
@@ -1,11 +1,13 @@
 """Utilities for creating deployments and related resources."""
 
-__all__ = ["get_kafka_bootstrap_server", "create_deployment", "create_service"]
+from __future__ import annotations
 
-from typing import Any, Dict, Mapping
+from typing import Any, Dict, Mapping, Optional
 
 import kopf
 
+__all__ = ["get_kafka_bootstrap_server", "create_deployment", "create_service"]
+
 
 def get_kafka_bootstrap_server(kafka, *, listener_name):
     """Get the bootstrap server address for a Strimzi Kafka cluster
@@ -121,7 +123,21 @@ def _get_v1beta1_bootstrap_server(
     )
 
 
-def create_deployment(*, name, bootstrap_server, secret_name, secret_version):
+def create_deployment(
+    *,
+    name: str,
+    bootstrap_server: str,
+    secret_name: str,
+    secret_version: str,
+    registry_image: str,
+    registry_image_tag: str,
+    registry_cpu_limit: Optional[str],
+    registry_cpu_request: Optional[str],
+    registry_mem_limit: Optional[str],
+    registry_mem_request: Optional[str],
+    compatibility_level: str,
+    security_protocol: str,
+) -> Dict[str, Any]:
     """Create the JSON resource for a Deployment of the Confluence Schema
     Registry.
 
@@ -139,6 +155,29 @@ def create_deployment(*, name, bootstrap_server, secret_name, secret_version):
     secret_version : `str`
         The ``resourceVersion`` of the Secret containing the JKS-formatted
         keystore and truststore.
+    registry_image : `str`
+        The Schema Registry docker image.
+    registry_image_tag : `str`
+        The tag for the Schema Registry docker image.
+    registry_cpu_limit : `str` or `None`
+        Requested CPU limit for the registry container. `None` omits the
+        setting from the container spec.
+    registry_cpu_request : `str` or `None`
+        Requested CPU allocation for the registry container. `None` omits the
+        setting from the container spec.
+    registry_mem_limit : `str` or `None`
+        Requested memory limit for the registry container. `None` omits the
+        setting from the container spec.
+    registry_mem_request : `str` or `None`
+        Requested memory allocation for the registry container. `None` omits
+        the setting from the container spec.
+    compatiblity_level : `str`
+        The default schema compatiblity in a subject. Can be one of:
+        none, backward, backward_transitive, forward, forward_transitive,
+        full, full_transitive.
+    security_protocol : `str`
+        The Kafka store security policy. Can be SSL, PLAINTEXT, SASL_PLAINTEXT,
+        or SASL_SSL.
 
     Returns
     -------
@@ -148,7 +187,16 @@ def create_deployment(*, name, bootstrap_server, secret_name, secret_version):
     key_prefix = "strimziregistryoperator.roundtable.lsst.codes"
 
     registry_container = create_container_spec(
-        secret_name=secret_name, bootstrap_server=bootstrap_server
+        secret_name=secret_name,
+        bootstrap_server=bootstrap_server,
+        registry_image=registry_image,
+        registry_image_tag=registry_image_tag,
+        registry_cpu_limit=registry_cpu_limit,
+        registry_cpu_request=registry_cpu_request,
+        registry_mem_limit=registry_mem_limit,
+        registry_mem_request=registry_mem_request,
+        compatibility_level=compatibility_level,
+        security_protocol=security_protocol,
     )
 
     # The pod template
@@ -179,8 +227,53 @@ def create_deployment(*, name, bootstrap_server, secret_name, secret_version):
     return dep
 
 
-def create_container_spec(*, secret_name, bootstrap_server):
-    """Create the container spec for the Schema Registry deployment."""
+def create_container_spec(
+    *,
+    secret_name: str,
+    bootstrap_server: str,
+    registry_image: str,
+    registry_image_tag: str,
+    registry_cpu_limit: Optional[str],
+    registry_cpu_request: Optional[str],
+    registry_mem_limit: Optional[str],
+    registry_mem_request: Optional[str],
+    compatibility_level: str,
+    security_protocol: str,
+) -> Dict[str, Any]:
+    """Create the container spec for the Schema Registry deployment.
+
+    Parameters
+    ----------
+    secret_name : `str`
+        Name of the Secret resource containing the JKS-formatted keystore
+        and truststore.
+    secret_version : `str`
+        The ``resourceVersion`` of the Secret containing the JKS-formatted
+        keystore and truststore.
+    registry_image : `str`
+        The Schema Registry docker image.
+    registry_image_tag : `str`
+        The tag for the Schema Registry docker image.
+    registry_cpu_limit : `str` or `None`
+        Requested CPU limit for the registry container. `None` omits the
+        setting from the container spec.
+    registry_cpu_request : `str` or `None`
+        Requested CPU allocation for the registry container. `None` omits the
+        setting from the container spec.
+    registry_mem_limit : `str` or `None`
+        Requested memory limit for the registry container. `None` omits the
+        setting from the container spec.
+    registry_mem_request : `str` or `None`
+        Requested memory allocation for the registry container. `None` omits
+        the setting from the container spec.
+    compatiblity_level : `str`
+        The default schema compatiblity in a subject. Can be one of:
+        none, backward, backward_transitive, forward, forward_transitive,
+        full, full_transitive.
+    security_protocol : `str`
+        The Kafka store security policy. Can be SSL, PLAINTEXT, SASL_PLAINTEXT,
+        or SASL_SSL.
+    """
     registry_env = [
         {
             "name": "SCHEMA_REGISTRY_HOST_NAME",
@@ -191,14 +284,9 @@ def create_container_spec(*, secret_name, bootstrap_server):
             "name": "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
             "value": bootstrap_server,
         },
-        # NOTE: This can likely be left to the default
-        # {
-        #     'name': 'SCHEMA_REGISTRY_KAFKASTORE_GROUP_ID',
-        #     'value': None,  # FIXME
-        # },
         {
-            "name": "SCHEMA_REGISTRY_AVRO_COMPATIBILITY_LEVEL",
-            "value": "forward",
+            "name": "SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL",
+            "value": compatibility_level,
         },
         {"name": "SCHEMA_REGISTRY_MASTER_ELIGIBILITY", "value": "true"},
         {
@@ -237,13 +325,13 @@ def create_container_spec(*, secret_name, bootstrap_server):
         },
         {
             "name": "SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL",
-            "value": "SSL",
+            "value": security_protocol,
         },
     ]
 
     registry_container = {
         "name": "server",
-        "image": "confluentinc/cp-schema-registry:5.3.1",
+        "image": f"{registry_image}:{registry_image_tag}",
         "imagePullPolicy": "IfNotPresent",
         "ports": [
             {
@@ -262,6 +350,29 @@ def create_container_spec(*, secret_name, bootstrap_server):
         ],
     }
 
+    if (
+        registry_cpu_limit
+        or registry_cpu_request
+        or registry_mem_limit
+        or registry_mem_request
+    ):
+        resource_spec: Dict[str, Dict[str, str]] = {}
+        if registry_cpu_limit or registry_mem_limit:
+            limit_spec: Dict[str, str] = {}
+            if registry_cpu_limit:
+                limit_spec["cpu"] = registry_cpu_limit
+            if registry_mem_limit:
+                limit_spec["memory"] = registry_mem_limit
+            resource_spec["limits"] = limit_spec
+        if registry_cpu_request or registry_mem_request:
+            request_spec: Dict[str, str] = {}
+            if registry_cpu_request:
+                request_spec["cpu"] = registry_cpu_request
+            if registry_mem_request:
+                request_spec["memory"] = registry_mem_request
+            resource_spec["requests"] = request_spec
+        registry_container["resources"] = resource_spec
+
     return registry_container
 
 
diff --git a/strimziregistryoperator/handlers/createregistry.py b/strimziregistryoperator/handlers/createregistry.py
index 69073aa..5199103 100644
--- a/strimziregistryoperator/handlers/createregistry.py
+++ b/strimziregistryoperator/handlers/createregistry.py
@@ -1,5 +1,9 @@
 """Kopf handler for the creation of a StrimziSchemaRegistry."""
 
+from __future__ import annotations
+
+from typing import Dict, Optional
+
 import kopf
 
 from .. import state
@@ -43,15 +47,22 @@ def create_registry(spec, meta, namespace, name, uid, logger, body, **kwargs):
 
     # Get configurations from StrimziSchemaRegistry
     try:
-        strimzi_api_version = spec["strimzi-version"]
+        strimzi_api_version = spec["strimziVersion"]
     except KeyError:
-        strimzi_api_version = "v1beta2"
-        logger.warning(
-            "StrimziSchemaRegistry %s is missing a strimzi-version, "
-            "using default %s",
-            name,
-            strimzi_api_version,
-        )
+        try:
+            strimzi_api_version = spec["strimzi-version"]
+            logger.warning(
+                "The strimzi-version configuration is deprecated. "
+                "Use strimziVersion instead."
+            )
+        except KeyError:
+            strimzi_api_version = "v1beta2"
+            logger.warning(
+                "StrimziSchemaRegistry %s is missing a strimziVersion, "
+                "using default %s",
+                name,
+                strimzi_api_version,
+            )
 
     try:
         listener_name = spec["listener"]
@@ -66,13 +77,34 @@ def create_registry(spec, meta, namespace, name, uid, logger, body, **kwargs):
 
     service_type = spec.get("serviceType", "ClusterIP")
 
+    registry_image = spec.get(
+        "registryImage", "confluentinc/cp-schema-registry"
+    )
+
+    registry_image_tag = spec.get("registryImageTag", "7.2.1")
+
+    # Limits and requests can be None so that they are omitted from the
+    # registry's deployment specification
+    registry_cpu_limit = get_nullable(spec, "cpuLimit")
+    registry_cpu_request = get_nullable(spec, "cpuRequest")
+    registry_mem_limit = get_nullable(spec, "memoryLimit")
+    registry_mem_request = get_nullable(spec, "memoryRequest")
+
+    # Additional schema Registry configurations
+    registry_compatibility_level = spec.get("compatibilitylevel", "forward")
+    security_protocol = spec.get("securityProtocol", "SSL")
+
     logger.info(
-        "Creating a new Schema Registry deployment: %s with listener=%s and "
-        "strimzi-version=%s serviceType=%s",
+        "Creating a new Schema Registry deployment: %s with listener=%s "
+        "(security protocol=%s) and strimzi-version=%s serviceType=%s "
+        "image=%s:%s",
         name,
         listener_name,
+        security_protocol,
         strimzi_api_version,
         service_type,
+        registry_image,
+        registry_image_tag,
     )
 
     # Get the name of the Kafka cluster associated with the
@@ -133,6 +165,14 @@ def create_registry(spec, meta, namespace, name, uid, logger, body, **kwargs):
             bootstrap_server=bootstrap_server,
             secret_name=secret_name,
             secret_version=secret_version,
+            registry_image=registry_image,
+            registry_image_tag=registry_image_tag,
+            registry_cpu_limit=registry_cpu_limit,
+            registry_cpu_request=registry_cpu_request,
+            registry_mem_limit=registry_mem_limit,
+            registry_mem_request=registry_mem_request,
+            compatibility_level=registry_compatibility_level,
+            security_protocol=security_protocol,
         )
         # Set the StrimziSchemaRegistry as the owner
         kopf.adopt(dep_body, owner=body)
@@ -163,3 +203,18 @@ def create_registry(spec, meta, namespace, name, uid, logger, body, **kwargs):
 
     # Add the name of the registry to the cache
     state.registry_names.add(name)
+
+
+def get_nullable(spec: Dict[str, str], key: str) -> Optional[str]:
+    """Get a nullable property of the StrimziSchemaRegistry resource.
+
+    If the propety is an empty string, it is null. If it is missing, it is =
+    null.
+    """
+    value = spec.get(key)
+    if value is None:
+        return None
+    elif value == "":
+        return None
+    else:
+        return value


Test patch:
diff --git a/tests/test_deployments.py b/tests/test_deployments.py
index 42b5bbf..7162f9d 100644
--- a/tests/test_deployments.py
+++ b/tests/test_deployments.py
@@ -1,10 +1,15 @@
 """Tests for the strimziregistryoperator.deployments module."""
 
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
 import kopf
 import pytest
 import yaml
 
 from strimziregistryoperator.deployments import (
+    create_deployment,
     create_service,
     get_kafka_bootstrap_server,
 )
@@ -165,3 +170,80 @@ def test_create_nodeport_service() -> None:
         name="confluent-schema-registry", service_type="NodePort"
     )
     assert resource["spec"]["type"] == "NodePort"
+
+
+def get_env_value(env: List[Dict[str, str]], name: str) -> Optional[str]:
+    """Get the value of an environment variable in the container spec.env"""
+    for item in env:
+        if item["name"] == name:
+            return item["value"]
+    return None
+
+
+def test_create_deployment_configurations() -> None:
+    """Create a schema registry deployment body with configurations."""
+    registry_image = "demo/testimage"
+    registry_image_tag = "1.2.3"
+
+    dep_body = create_deployment(
+        name="example-server",
+        bootstrap_server="example-server.default.svc:9093",
+        secret_name="example-server",
+        secret_version="1",
+        registry_image=registry_image,
+        registry_image_tag=registry_image_tag,
+        registry_cpu_limit=None,
+        registry_mem_limit=None,
+        registry_cpu_request=None,
+        registry_mem_request=None,
+        compatibility_level="backward",
+        security_protocol="SSL",
+    )
+    assert dep_body["spec"]["template"]["spec"]["containers"][0]["image"] == (
+        f"{registry_image}:{registry_image_tag}"
+    )
+
+    env = dep_body["spec"]["template"]["spec"]["containers"][0]["env"]
+    assert (
+        get_env_value(env, "SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL")
+        == "backward"
+    )
+    assert (
+        get_env_value(env, "SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL")
+        == "SSL"
+    )
+
+    # no resource settings
+    assert (
+        "resources"
+        not in dep_body["spec"]["template"]["spec"]["containers"][0]
+    )
+
+
+def test_create_deployment_resource_settings() -> None:
+    """Create a schema registry deployment body with a customized image."""
+    registry_image = "demo/testimage"
+    registry_image_tag = "1.2.3"
+
+    dep_body = create_deployment(
+        name="example-server",
+        bootstrap_server="example-server.default.svc:9093",
+        secret_name="example-server",
+        secret_version="1",
+        registry_image=registry_image,
+        registry_image_tag=registry_image_tag,
+        registry_cpu_limit="1000m",
+        registry_mem_limit="1000M",
+        registry_cpu_request="100m",
+        registry_mem_request="768M",
+        compatibility_level="forward",
+        security_protocol="SSL",
+    )
+    print(dep_body)
+    resources = dep_body["spec"]["template"]["spec"]["containers"][0][
+        "resources"
+    ]
+    assert resources["limits"]["cpu"] == "1000m"
+    assert resources["limits"]["memory"] == "1000M"
+    assert resources["requests"]["cpu"] == "100m"
+    assert resources["requests"]["memory"] == "768M"