Skip to main content
krabka
GitHub
All operator CRDs

KafkaConnector

Expand a field to see its description, example, and default behavior. Nested fields are grouped under their parent.

API group
krabka.io
Scope
Namespaced
Resource name
kafkaconnectors
Short names
kc

Required fields must be present when their parent object exists. [] denotes an array item; {key} denotes a map value. status is written by the operator. Reconcile-time checks and runtime defaults can add requirements beyond the schema.

Examples are illustrative JSON values. Open “View JSON example” for an object or array. Adapt names, endpoints, and credentials to your cluster. Kubernetes applies schema defaults when the containing object exists. Runtime and omission notes describe operator behavior. Status examples show reported data.

Runtime notes and domain examples were reviewed against operator source at b409b56.

View the complete CRD manifest at 618d621
Example manifest

This example includes the required spec fields. Adapt its values and create any referenced resources before applying it.

apiVersion: krabka.io/v1alpha1
kind: KafkaConnector
metadata:
  name: example
  namespace: default
  labels:
    krabka.io/cluster: my-cluster
spec:
  databaseUrl:
    name: orders-database
    key: url
  publication: krabka_connect
  schemaRegistryUrl: http://schema-registry.kafka.svc.cluster.local:8081
  slot: orders_krabka
  type: postgresSource

krabka.io/v1alpha1

Served API version · Storage version · 39 schema entries

Complete OpenAPI schema (JSON) 
{
  "description": "Auto-generated derived type for KafkaConnectorSpec via `CustomResource`",
  "properties": {
    "spec": {
      "properties": {
        "databaseUrl": {
          "description": "`Secret` key containing the `PostgreSQL` connection URL.",
          "properties": {
            "key": {
              "description": "Key containing the `PostgreSQL` connection URL.",
              "type": "string"
            },
            "name": {
              "description": "Secret name.",
              "type": "string"
            }
          },
          "required": [
            "key",
            "name"
          ],
          "type": "object"
        },
        "image": {
          "description": "Container image override. When absent, the operator default is used.",
          "nullable": true,
          "type": "string"
        },
        "paused": {
          "default": false,
          "description": "Gracefully stop the worker while retaining its durable checkpoint.",
          "type": "boolean"
        },
        "publication": {
          "description": "`PostgreSQL` publication read by the connector.",
          "minLength": 1,
          "type": "string"
        },
        "resources": {
          "description": "Worker `CPU` and memory requests/limits.",
          "nullable": true,
          "properties": {
            "claims": {
              "description": "Claims lists the names of resources, defined in spec.resourceClaims, that are used by this container.\n\nThis is an alpha field and requires enabling the DynamicResourceAllocation feature gate.\n\nThis field is immutable. It can only be set for containers.",
              "items": {
                "description": "ResourceClaim references one entry in PodSpec.ResourceClaims.",
                "properties": {
                  "name": {
                    "description": "Name must match the name of one entry in pod.spec.resourceClaims of the Pod where this field is used. It makes that resource available inside a container.",
                    "type": "string"
                  },
                  "request": {
                    "description": "Request is the name chosen for a request in the referenced claim. If empty, everything from the claim is made available, otherwise only the result of this request.",
                    "type": "string"
                  }
                },
                "required": [
                  "name"
                ],
                "type": "object"
              },
              "type": "array"
            },
            "limits": {
              "additionalProperties": {
                "description": "Quantity is a fixed-point representation of a number. It provides convenient marshaling/unmarshaling in JSON and YAML, in addition to String() and AsInt64() accessors.\n\nThe serialization format is:\n\n``` <quantity>        ::= <signedNumber><suffix>\n\n\t(Note that <suffix> may be empty, from the \"\" case in <decimalSI>.)\n\n<digit>           ::= 0 | 1 | ... | 9 <digits>          ::= <digit> | <digit><digits> <number>          ::= <digits> | <digits>.<digits> | <digits>. | .<digits> <sign>            ::= \"+\" | \"-\" <signedNumber>    ::= <number> | <sign><number> <suffix>          ::= <binarySI> | <decimalExponent> | <decimalSI> <binarySI>        ::= Ki | Mi | Gi | Ti | Pi | Ei\n\n\t(International System of units; See: http://physics.nist.gov/cuu/Units/binary.html)\n\n<decimalSI>       ::= m | \"\" | k | M | G | T | P | E\n\n\t(Note that 1024 = 1Ki but 1000 = 1k; I didn't choose the capitalization.)\n\n<decimalExponent> ::= \"e\" <signedNumber> | \"E\" <signedNumber> ```\n\nNo matter which of the three exponent forms is used, no quantity may represent a number greater than 2^63-1 in magnitude, nor may it have more than 3 decimal places. Numbers larger or more precise will be capped or rounded up. (E.g.: 0.1m will rounded up to 1m.) This may be extended in the future if we require larger or smaller quantities.\n\nWhen a Quantity is parsed from a string, it will remember the type of suffix it had, and will use the same type again when it is serialized.\n\nBefore serializing, Quantity will be put in \"canonical form\". This means that Exponent/suffix will be adjusted up or down (with a corresponding increase or decrease in Mantissa) such that:\n\n- No precision is lost - No fractional digits will be emitted - The exponent (or suffix) is as large as possible.\n\nThe sign will be omitted unless the number is negative.\n\nExamples:\n\n- 1.5 will be serialized as \"1500m\" - 1.5Gi will be serialized as \"1536Mi\"\n\nNote that the quantity will NEVER be internally represented by a floating point number. That is the whole point of this exercise.\n\nNon-canonical values will still parse as long as they are well formed, but will be re-emitted in their canonical form. (So always use canonical form, or don't diff.)\n\nThis format is intended to make it difficult to use these numbers without writing some sort of special handling code in the hopes that that will cause implementors to also use a fixed point implementation.",
                "x-kubernetes-int-or-string": true
              },
              "description": "Limits describes the maximum amount of compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/",
              "type": "object"
            },
            "requests": {
              "additionalProperties": {
                "description": "Quantity is a fixed-point representation of a number. It provides convenient marshaling/unmarshaling in JSON and YAML, in addition to String() and AsInt64() accessors.\n\nThe serialization format is:\n\n``` <quantity>        ::= <signedNumber><suffix>\n\n\t(Note that <suffix> may be empty, from the \"\" case in <decimalSI>.)\n\n<digit>           ::= 0 | 1 | ... | 9 <digits>          ::= <digit> | <digit><digits> <number>          ::= <digits> | <digits>.<digits> | <digits>. | .<digits> <sign>            ::= \"+\" | \"-\" <signedNumber>    ::= <number> | <sign><number> <suffix>          ::= <binarySI> | <decimalExponent> | <decimalSI> <binarySI>        ::= Ki | Mi | Gi | Ti | Pi | Ei\n\n\t(International System of units; See: http://physics.nist.gov/cuu/Units/binary.html)\n\n<decimalSI>       ::= m | \"\" | k | M | G | T | P | E\n\n\t(Note that 1024 = 1Ki but 1000 = 1k; I didn't choose the capitalization.)\n\n<decimalExponent> ::= \"e\" <signedNumber> | \"E\" <signedNumber> ```\n\nNo matter which of the three exponent forms is used, no quantity may represent a number greater than 2^63-1 in magnitude, nor may it have more than 3 decimal places. Numbers larger or more precise will be capped or rounded up. (E.g.: 0.1m will rounded up to 1m.) This may be extended in the future if we require larger or smaller quantities.\n\nWhen a Quantity is parsed from a string, it will remember the type of suffix it had, and will use the same type again when it is serialized.\n\nBefore serializing, Quantity will be put in \"canonical form\". This means that Exponent/suffix will be adjusted up or down (with a corresponding increase or decrease in Mantissa) such that:\n\n- No precision is lost - No fractional digits will be emitted - The exponent (or suffix) is as large as possible.\n\nThe sign will be omitted unless the number is negative.\n\nExamples:\n\n- 1.5 will be serialized as \"1500m\" - 1.5Gi will be serialized as \"1536Mi\"\n\nNote that the quantity will NEVER be internally represented by a floating point number. That is the whole point of this exercise.\n\nNon-canonical values will still parse as long as they are well formed, but will be re-emitted in their canonical form. (So always use canonical form, or don't diff.)\n\nThis format is intended to make it difficult to use these numbers without writing some sort of special handling code in the hopes that that will cause implementors to also use a fixed point implementation.",
                "x-kubernetes-int-or-string": true
              },
              "description": "Requests describes the minimum amount of compute resources required. If Requests is omitted for a container, it defaults to Limits if that is explicitly specified, otherwise to an implementation-defined value. Requests cannot exceed Limits. More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/",
              "type": "object"
            }
          },
          "type": "object"
        },
        "runtime": {
          "description": "Optional worker runtime overrides.",
          "nullable": true,
          "properties": {
            "commitIntervalMs": {
              "description": "Maximum interval between durable checkpoint commits, in milliseconds.",
              "format": "uint64",
              "minimum": 1,
              "nullable": true,
              "type": "integer"
            },
            "maxBatch": {
              "description": "Maximum records read in one source poll. Worker default: 500.",
              "format": "uint",
              "minimum": 1,
              "nullable": true,
              "type": "integer"
            },
            "pollBackoffMs": {
              "description": "Delay after an empty source poll, in milliseconds.",
              "format": "uint64",
              "minimum": 1,
              "nullable": true,
              "type": "integer"
            }
          },
          "type": "object"
        },
        "schema": {
          "description": "Default `PostgreSQL` schema for unqualified table names. Default `public`.",
          "nullable": true,
          "type": "string"
        },
        "schemaRegistryUrl": {
          "description": "Confluent-compatible Schema Registry base URL used to allocate record schema IDs.",
          "minLength": 1,
          "type": "string"
        },
        "slot": {
          "description": "`PostgreSQL` logical replication slot.",
          "minLength": 1,
          "type": "string"
        },
        "tables": {
          "description": "Unqualified tables included in CDC. All belong to `schema`.",
          "items": {
            "type": "string"
          },
          "minItems": 1,
          "type": "array"
        },
        "topicPrefix": {
          "description": "Prefix prepended to destination topic names.",
          "nullable": true,
          "type": "string"
        },
        "type": {
          "description": "Connector implementation. The initial release supports `postgresSource`.",
          "enum": [
            "postgresSource"
          ],
          "type": "string"
        }
      },
      "required": [
        "databaseUrl",
        "publication",
        "schemaRegistryUrl",
        "slot",
        "type"
      ],
      "type": "object"
    },
    "status": {
      "description": "Observed state for an operator-managed connector.",
      "nullable": true,
      "properties": {
        "conditions": {
          "default": [],
          "description": "Standard `Ready`, `Paused`, and `Failed` conditions.",
          "items": {
            "properties": {
              "lastTransitionTime": {
                "description": "RFC3339 timestamp.",
                "type": "string"
              },
              "message": {
                "description": "Human-readable message.",
                "type": "string"
              },
              "reason": {
                "description": "CamelCase machine reason.",
                "type": "string"
              },
              "status": {
                "description": "`True`, `False`, or `Unknown`.",
                "type": "string"
              },
              "type": {
                "description": "For example, `Ready`.",
                "type": "string"
              }
            },
            "required": [
              "lastTransitionTime",
              "message",
              "reason",
              "status",
              "type"
            ],
            "type": "object"
          },
          "type": "array"
        },
        "observedGeneration": {
          "description": "Generation represented by the rendered Deployment.",
          "format": "int64",
          "nullable": true,
          "type": "integer"
        },
        "readyReplicas": {
          "description": "Ready worker replicas reported by the Deployment.",
          "format": "int32",
          "nullable": true,
          "type": "integer"
        },
        "replicas": {
          "description": "Desired worker replicas (zero while paused, otherwise one).",
          "format": "int32",
          "nullable": true,
          "type": "integer"
        }
      },
      "type": "object"
    }
  },
  "required": [
    "spec"
  ],
  "title": "KafkaConnector",
  "type": "object"
}

spec

spec object Required 12 fields
Example
View JSON example 
{
  "databaseUrl": {
    "name": "orders-database",
    "key": "url"
  },
  "publication": "krabka_connect",
  "schemaRegistryUrl": "http://schema-registry.kafka.svc.cluster.local:8081",
  "slot": "orders_krabka",
  "type": "postgresSource"
}
Default No schema default
Not declared
databaseUrl object Required 2 fields

Secret key containing the PostgreSQL connection URL.

Example
View JSON example 
{
  "name": "orders-database",
  "key": "url"
}
Default No schema default
Not declared
key string Required

Key containing the PostgreSQL connection URL.

Example
"url"
Default No schema default
Not declared
#
name string Required

Secret name.

Example
"orders-database"
Default No schema default
Not declared
#
#
image string | null Optional

Container image override. When absent, the operator default is used.

Example
"registry.example.com/krabka/connector:1.0.0"
Default Runtime / omission

Use the operator's --default-connector-image configuration when omitted.

#
paused boolean Optional

Gracefully stop the worker while retaining its durable checkpoint.

Example
false
Default Schema default
false
#
publication string Required

PostgreSQL publication read by the connector.

Example
"krabka_connect"
Default No schema default
Not declared
Validation and schema details
Minimum length
1
#
resources object | null Optional 3 fields

Worker CPU and memory requests/limits.

Example
View JSON example 
{
  "claims": [
    {
      "name": "example"
    }
  ]
}
Default No schema default
Not declared
claims array Optional 1 field

Claims lists the names of resources, defined in spec.resourceClaims, that are used by this container.

This is an alpha field and requires enabling the DynamicResourceAllocation feature gate.

This field is immutable. It can only be set for containers.

Example
View JSON example 
[
  {
    "name": "example"
  }
]
Default No schema default
Not declared
[] object Array item 2 fields

ResourceClaim references one entry in PodSpec.ResourceClaims.

Example
View JSON example 
{
  "name": "example"
}
Default No independent default
Defined by the parent
name string Required

Name must match the name of one entry in pod.spec.resourceClaims of the Pod where this field is used. It makes that resource available inside a container.

Example
"example"
Default No schema default
Not declared
#
request string Optional

Request is the name chosen for a request in the referenced claim. If empty, everything from the claim is made available, otherwise only the result of this request.

Example
"example"
Default No schema default
Not declared
#
#
#
limits object Optional 1 field

Limits describes the maximum amount of compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/

Example
View JSON example 
{
  "example": "256Mi"
}
Default No schema default
Not declared
Validation and schema details
Additional properties
{
  "description": "Quantity is a fixed-point representation of a number. It provides convenient marshaling/unmarshaling in JSON and YAML, in addition to String() and AsInt64() accessors.\n\nThe serialization format is:\n\n``` <quantity>        ::= <signedNumber><suffix>\n\n\t(Note that <suffix> may be empty, from the \"\" case in <decimalSI>.)\n\n<digit>           ::= 0 | 1 | ... | 9 <digits>          ::= <digit> | <digit><digits> <number>          ::= <digits> | <digits>.<digits> | <digits>. | .<digits> <sign>            ::= \"+\" | \"-\" <signedNumber>    ::= <number> | <sign><number> <suffix>          ::= <binarySI> | <decimalExponent> | <decimalSI> <binarySI>        ::= Ki | Mi | Gi | Ti | Pi | Ei\n\n\t(International System of units; See: http://physics.nist.gov/cuu/Units/binary.html)\n\n<decimalSI>       ::= m | \"\" | k | M | G | T | P | E\n\n\t(Note that 1024 = 1Ki but 1000 = 1k; I didn't choose the capitalization.)\n\n<decimalExponent> ::= \"e\" <signedNumber> | \"E\" <signedNumber> ```\n\nNo matter which of the three exponent forms is used, no quantity may represent a number greater than 2^63-1 in magnitude, nor may it have more than 3 decimal places. Numbers larger or more precise will be capped or rounded up. (E.g.: 0.1m will rounded up to 1m.) This may be extended in the future if we require larger or smaller quantities.\n\nWhen a Quantity is parsed from a string, it will remember the type of suffix it had, and will use the same type again when it is serialized.\n\nBefore serializing, Quantity will be put in \"canonical form\". This means that Exponent/suffix will be adjusted up or down (with a corresponding increase or decrease in Mantissa) such that:\n\n- No precision is lost - No fractional digits will be emitted - The exponent (or suffix) is as large as possible.\n\nThe sign will be omitted unless the number is negative.\n\nExamples:\n\n- 1.5 will be serialized as \"1500m\" - 1.5Gi will be serialized as \"1536Mi\"\n\nNote that the quantity will NEVER be internally represented by a floating point number. That is the whole point of this exercise.\n\nNon-canonical values will still parse as long as they are well formed, but will be re-emitted in their canonical form. (So always use canonical form, or don't diff.)\n\nThis format is intended to make it difficult to use these numbers without writing some sort of special handling code in the hopes that that will cause implementors to also use a fixed point implementation.",
  "x-kubernetes-int-or-string": true
}
{key} integer | string Map value

Quantity is a fixed-point representation of a number. It provides convenient marshaling/unmarshaling in JSON and YAML, in addition to String() and AsInt64() accessors.

The serialization format is:

<quantity>        ::= <signedNumber><suffix>

	(Note that <suffix> may be empty, from the "" case in <decimalSI>.)

<digit>           ::= 0 | 1 | ... | 9 <digits>          ::= <digit> | <digit><digits> <number>          ::= <digits> | <digits>.<digits> | <digits>. | .<digits> <sign>            ::= "+" | "-" <signedNumber>    ::= <number> | <sign><number> <suffix>          ::= <binarySI> | <decimalExponent> | <decimalSI> <binarySI>        ::= Ki | Mi | Gi | Ti | Pi | Ei

	(International System of units; See: http://physics.nist.gov/cuu/Units/binary.html)

<decimalSI>       ::= m | "" | k | M | G | T | P | E

	(Note that 1024 = 1Ki but 1000 = 1k; I didn't choose the capitalization.)

<decimalExponent> ::= "e" <signedNumber> | "E" <signedNumber>

No matter which of the three exponent forms is used, no quantity may represent a number greater than 2^63-1 in magnitude, nor may it have more than 3 decimal places. Numbers larger or more precise will be capped or rounded up. (E.g.: 0.1m will rounded up to 1m.) This may be extended in the future if we require larger or smaller quantities.

When a Quantity is parsed from a string, it will remember the type of suffix it had, and will use the same type again when it is serialized.

Before serializing, Quantity will be put in "canonical form". This means that Exponent/suffix will be adjusted up or down (with a corresponding increase or decrease in Mantissa) such that:

  • No precision is lost - No fractional digits will be emitted - The exponent (or suffix) is as large as possible.

The sign will be omitted unless the number is negative.

Examples:

  • 1.5 will be serialized as "1500m" - 1.5Gi will be serialized as "1536Mi"

Note that the quantity will NEVER be internally represented by a floating point number. That is the whole point of this exercise.

Non-canonical values will still parse as long as they are well formed, but will be re-emitted in their canonical form. (So always use canonical form, or don't diff.)

This format is intended to make it difficult to use these numbers without writing some sort of special handling code in the hopes that that will cause implementors to also use a fixed point implementation.

Example
"256Mi"
Default No independent default
Defined by the parent
Validation and schema details
x-kubernetes-int-or-string
true
#
#
requests object Optional 1 field

Requests describes the minimum amount of compute resources required. If Requests is omitted for a container, it defaults to Limits if that is explicitly specified, otherwise to an implementation-defined value. Requests cannot exceed Limits. More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/

Example
View JSON example 
{
  "example": "256Mi"
}
Default Documented behavior

If Requests is omitted for a container, it defaults to Limits if that is explicitly specified, otherwise to an implementation-defined value.

Validation and schema details
Additional properties
{
  "description": "Quantity is a fixed-point representation of a number. It provides convenient marshaling/unmarshaling in JSON and YAML, in addition to String() and AsInt64() accessors.\n\nThe serialization format is:\n\n``` <quantity>        ::= <signedNumber><suffix>\n\n\t(Note that <suffix> may be empty, from the \"\" case in <decimalSI>.)\n\n<digit>           ::= 0 | 1 | ... | 9 <digits>          ::= <digit> | <digit><digits> <number>          ::= <digits> | <digits>.<digits> | <digits>. | .<digits> <sign>            ::= \"+\" | \"-\" <signedNumber>    ::= <number> | <sign><number> <suffix>          ::= <binarySI> | <decimalExponent> | <decimalSI> <binarySI>        ::= Ki | Mi | Gi | Ti | Pi | Ei\n\n\t(International System of units; See: http://physics.nist.gov/cuu/Units/binary.html)\n\n<decimalSI>       ::= m | \"\" | k | M | G | T | P | E\n\n\t(Note that 1024 = 1Ki but 1000 = 1k; I didn't choose the capitalization.)\n\n<decimalExponent> ::= \"e\" <signedNumber> | \"E\" <signedNumber> ```\n\nNo matter which of the three exponent forms is used, no quantity may represent a number greater than 2^63-1 in magnitude, nor may it have more than 3 decimal places. Numbers larger or more precise will be capped or rounded up. (E.g.: 0.1m will rounded up to 1m.) This may be extended in the future if we require larger or smaller quantities.\n\nWhen a Quantity is parsed from a string, it will remember the type of suffix it had, and will use the same type again when it is serialized.\n\nBefore serializing, Quantity will be put in \"canonical form\". This means that Exponent/suffix will be adjusted up or down (with a corresponding increase or decrease in Mantissa) such that:\n\n- No precision is lost - No fractional digits will be emitted - The exponent (or suffix) is as large as possible.\n\nThe sign will be omitted unless the number is negative.\n\nExamples:\n\n- 1.5 will be serialized as \"1500m\" - 1.5Gi will be serialized as \"1536Mi\"\n\nNote that the quantity will NEVER be internally represented by a floating point number. That is the whole point of this exercise.\n\nNon-canonical values will still parse as long as they are well formed, but will be re-emitted in their canonical form. (So always use canonical form, or don't diff.)\n\nThis format is intended to make it difficult to use these numbers without writing some sort of special handling code in the hopes that that will cause implementors to also use a fixed point implementation.",
  "x-kubernetes-int-or-string": true
}
{key} integer | string Map value

Quantity is a fixed-point representation of a number. It provides convenient marshaling/unmarshaling in JSON and YAML, in addition to String() and AsInt64() accessors.

The serialization format is:

<quantity>        ::= <signedNumber><suffix>

	(Note that <suffix> may be empty, from the "" case in <decimalSI>.)

<digit>           ::= 0 | 1 | ... | 9 <digits>          ::= <digit> | <digit><digits> <number>          ::= <digits> | <digits>.<digits> | <digits>. | .<digits> <sign>            ::= "+" | "-" <signedNumber>    ::= <number> | <sign><number> <suffix>          ::= <binarySI> | <decimalExponent> | <decimalSI> <binarySI>        ::= Ki | Mi | Gi | Ti | Pi | Ei

	(International System of units; See: http://physics.nist.gov/cuu/Units/binary.html)

<decimalSI>       ::= m | "" | k | M | G | T | P | E

	(Note that 1024 = 1Ki but 1000 = 1k; I didn't choose the capitalization.)

<decimalExponent> ::= "e" <signedNumber> | "E" <signedNumber>

No matter which of the three exponent forms is used, no quantity may represent a number greater than 2^63-1 in magnitude, nor may it have more than 3 decimal places. Numbers larger or more precise will be capped or rounded up. (E.g.: 0.1m will rounded up to 1m.) This may be extended in the future if we require larger or smaller quantities.

When a Quantity is parsed from a string, it will remember the type of suffix it had, and will use the same type again when it is serialized.

Before serializing, Quantity will be put in "canonical form". This means that Exponent/suffix will be adjusted up or down (with a corresponding increase or decrease in Mantissa) such that:

  • No precision is lost - No fractional digits will be emitted - The exponent (or suffix) is as large as possible.

The sign will be omitted unless the number is negative.

Examples:

  • 1.5 will be serialized as "1500m" - 1.5Gi will be serialized as "1536Mi"

Note that the quantity will NEVER be internally represented by a floating point number. That is the whole point of this exercise.

Non-canonical values will still parse as long as they are well formed, but will be re-emitted in their canonical form. (So always use canonical form, or don't diff.)

This format is intended to make it difficult to use these numbers without writing some sort of special handling code in the hopes that that will cause implementors to also use a fixed point implementation.

Example
"256Mi"
Default No independent default
Defined by the parent
Validation and schema details
x-kubernetes-int-or-string
true
#
#
#
runtime object | null Optional 3 fields

Optional worker runtime overrides.

Example
View JSON example 
{
  "maxBatch": 500,
  "commitIntervalMs": 1000,
  "pollBackoffMs": 100
}
Default No schema default
Not declared
commitIntervalMs integer | null (uint64) Optional

Maximum interval between durable checkpoint commits, in milliseconds.

Example
1000
Default No schema default
Not declared
Validation and schema details
Minimum
1
#
maxBatch integer | null (uint) Optional

Maximum records read in one source poll. Worker default: 500.

Example
500
Default Documented behavior

Worker default: 500.

Validation and schema details
Minimum
1
#
pollBackoffMs integer | null (uint64) Optional

Delay after an empty source poll, in milliseconds.

Example
100
Default No schema default
Not declared
Validation and schema details
Minimum
1
#
#
schema string | null Optional

Default PostgreSQL schema for unqualified table names. Default public.

Example
"public"
Default Documented behavior

Default PostgreSQL schema for unqualified table names. Default public.

#
schemaRegistryUrl string Required

Confluent-compatible Schema Registry base URL used to allocate record schema IDs.

Example
"http://schema-registry.kafka.svc.cluster.local:8081"
Default No schema default
Not declared
Validation and schema details
Minimum length
1
#
slot string Required

PostgreSQL logical replication slot.

Example
"orders_krabka"
Default No schema default
Not declared
Validation and schema details
Minimum length
1
#
tables array Optional 1 field

Unqualified tables included in CDC. All belong to schema.

Example
View JSON example 
[
  "orders"
]
Default Runtime / omission

[] when omitted. If present, the list must contain at least one table.

Validation and schema details
Minimum items
1
[] string Array item
Example
"orders"
Default No independent default
Defined by the parent
#
#
topicPrefix string | null Optional

Prefix prepended to destination topic names.

Example
"db"
Default No schema default
Not declared
#
type string Required

Connector implementation. The initial release supports postgresSource.

Example
"postgresSource"
Default No schema default
Not declared
Allowed values"postgresSource"
#
#

status

status object | null Optional 4 fields

Observed state for an operator-managed connector.

Example
View JSON example 
{
  "conditions": [
    {
      "lastTransitionTime": "2026-10-05T12:00:00Z",
      "message": "The resource is ready.",
      "reason": "Reconciled",
      "status": "True",
      "type": "Ready"
    }
  ]
}
Default No schema default
Not declared
conditions array Optional 1 field

Standard Ready, Paused, and Failed conditions.

Example
View JSON example 
[
  {
    "lastTransitionTime": "2026-10-05T12:00:00Z",
    "message": "The resource is ready.",
    "reason": "Reconciled",
    "status": "True",
    "type": "Ready"
  }
]
Default Schema default
[]
[] object Array item 5 fields
Example
View JSON example 
{
  "lastTransitionTime": "2026-10-05T12:00:00Z",
  "message": "The resource is ready.",
  "reason": "Reconciled",
  "status": "True",
  "type": "Ready"
}
Default No schema default
Not declared
lastTransitionTime string Required

RFC3339 timestamp.

Example
"2026-10-05T12:00:00Z"
Default No schema default
Not declared
#
message string Required

Human-readable message.

Example
"The resource is ready."
Default No schema default
Not declared
#
reason string Required

CamelCase machine reason.

Example
"Reconciled"
Default No schema default
Not declared
#
status string Required

True, False, or Unknown.

Example
"True"
Default No schema default
Not declared
#
type string Required

For example, Ready.

Example
"Ready"
Default No schema default
Not declared
#
#
#
observedGeneration integer | null (int64) Optional

Generation represented by the rendered Deployment.

Example
1
Default No schema default
Not declared
#
readyReplicas integer | null (int32) Optional

Ready worker replicas reported by the Deployment.

Example
1
Default No schema default
Not declared
#
replicas integer | null (int32) Optional

Desired worker replicas (zero while paused, otherwise one).

Example
1
Default No schema default
Not declared
#
#

Resource schema

Resource schema object Resource schema

Auto-generated derived type for KafkaConnectorSpec via CustomResource

Example
View JSON example 
{
  "apiVersion": "krabka.io/v1alpha1",
  "kind": "KafkaConnector",
  "metadata": {
    "name": "example",
    "namespace": "default",
    "labels": {
      "krabka.io/cluster": "my-cluster"
    }
  },
  "spec": {
    "databaseUrl": {
      "name": "orders-database",
      "key": "url"
    },
    "publication": "krabka_connect",
    "schemaRegistryUrl": "http://schema-registry.kafka.svc.cluster.local:8081",
    "slot": "orders_krabka",
    "type": "postgresSource"
  }
}
Default No resource default
None
#