Chapter 3. Enhancements

For an overview of the enhancements introduced with Kafka 2.8.0, refer to the Kafka 2.8.0 Release Notes.

You can use build configuration so that AMQ Streams automatically builds a container image with the connector plugins you require for your data connections.

A dedicated service account is now created with Kafka Connect build pods. The service account is distinct from Kafka Connect itself. Before this release, the build ran under the default service account. Having its own identity is useful when specifying authentication and access.

Kafka Connect build now also works behind proxies if standard HTTP proxies (HTTP_PROXY, HTTPS_PROXY, and NO_PROXY) are set as environment variables for the AMQ Streams deployment.

See:

Use the Kubernetes Configuration Provider plugin to load configuration data from external sources. You can load data from OpenShift Secrets or ConfigMaps.

The provider operates independently of AMQ Streams. It loads the data without needing to restart the Kafka component, even when using a new Secret or ConfigMap.

You can use it to load configuration data for all Kafka components, including producers and consumers. Use it, for example, to provide the credentials for a Kafka Connect instance hosting multiple connectors without disruption

See Loading configuration values from external sources.

If you are using a ConfigMap to configure the (log4j2) logging levels for AMQ Streams Operators, you can now define logging filters to limit what is returned in the log. You add the filter properties to the ConfigMap.

The filters use markers to specify what to include in the log. You specify a kind, namespace and name for the marker. For example, if a Kafka cluster is failing, you can isolate the logs by specifying the kind as Kafka, and use the namespace and name of the failing cluster.

This example shows a marker filter for a Kafka cluster named my-kafka-cluster.

appender.console.filter.filter1.type=MarkerFilter 
appender.console.filter.filter1.onMatch=ACCEPT 
appender.console.filter.filter1.onMismatch=DENY 
appender.console.filter.filter1.marker=Kafka(my-namespace/my-kafka-cluster) 

appender.console.filter.filter1.type=MarkerFilter 

appender.console.filter.filter1.onMatch=ACCEPT 

appender.console.filter.filter1.onMismatch=DENY 

appender.console.filter.filter1.marker=Kafka(my-namespace/my-kafka-cluster) 

See Adding logging filters to Operators

Use these properties in the following scenarios:

These properties affect whether a client can obtain a token and the content of the token. They do not affect token validation rules imposed by the listener.

# ...
  authentication:
    type: oauth
    # ...
    clientAudience: AUDIENCE
    clientScope: SCOPE

# ...
  authentication:
    type: oauth
    # ...
    clientAudience: AUDIENCE
    clientScope: SCOPE

Authorization servers sometimes provide aud (audience) claims in JWT access tokens. When audience checks are enabled, the Kafka broker rejects tokens that do not contain the broker’s clientId in their aud claims. To enable audience checks, set the checkAudience option to true in the oauth listener configuration. Audience checks are disabled by default.

See Configuring OAuth 2.0 support for Kafka brokers and KafkaListenerAuthenticationOAuth schema reference

See KafkaClientAuthenticationOAuth schema reference

# ...
  authentication:
    type: oauth
    # ...
    enablePlain: true
    tokenEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/token

# ...
  authentication:
    type: oauth
    # ...
    enablePlain: true
    tokenEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/token

If the tokenEndpointUri is not specified, the listener treats the:

The behavior of the "long-lived access token" method for OAuth 2.0 over PLAIN authentication is unchanged. The tokenEndpointUri is not required when using this method.

See OAuth 2.0 authentication mechanisms

The handling of user quotas through the User Operator is no longer managed by ZooKeeper. Instead, user quotas are handled through the API.

Additionally, support has been added for Kafka’s mutation rate quota. This quota limits the number of partition mutations allowed per second. The quota prevents Kafka clusters from being overwhelmed by concurrent topic operations.

The number of partition mutations includes the following types of user requests:

You can configure a mutation rate quota to control the rate at which mutations are accepted for user requests. The rate is accumulated from the number of partitions created or deleted.

Use the controllerMutationRate option to apply the quota to the Kafka user. In this example, 10 partition creation and deletion operations are allowed per second.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  quotas:
    #...
    controllerMutationRate: 10 

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  quotas:
    #...
    controllerMutationRate: 10 

See User quotas

You can pause the reconciliation of custom resources managed by AMQ Streams operators to perform fixes or make updates. You can also pause reconciliation of custom resources you are creating. The custom resource is created, but it is ignored.

You add an annotation to the custom resource to pause it.

oc annotate KIND-OF-CUSTOM-RESOURCE NAME-OF-CUSTOM-RESOURCE strimzi.io/pause-reconciliation="true"

oc annotate KIND-OF-CUSTOM-RESOURCE NAME-OF-CUSTOM-RESOURCE strimzi.io/pause-reconciliation="true"

It is now possible to pause reconciliation of KafkaTopic custom resources.

See Pausing reconciliation of custom resources

The custom version of Kafka Exporter that is provided with AMQ Streams has been updated to version 1.3.1. AMQ Streams includes an example Grafana dashboard for Kafka Exporter in the examples provided (examples/metrics/grafana-dashboards/strimzi-kafka-exporter.json).

See Add Kafka Exporter

You can configure a KafkaConnect resource to create a custom Kafka Connect container image. Using spec.build configuration automates the process. You configure plugins to specify the implementation artifacts and output to reference the container registry that stores the image. AMQ Streams downloads and adds the connector plugins into the new container image.

The build process now uses the URL hash to name downloaded artifact files. Previously, it used the last segment of the download URL. If your plugin artifact requires a specific name, you can use a new other artifact type and its fileName field.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  build:
    output:
      #...
    plugins:
      - name: my-plugin
        artifacts:
          - type: other
            url: https://my-domain.tld/my-other-file.ext
            sha512sum: 589...ab4
            fileName: name-of-file.ext
  #...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
spec:
  #...
  build:
    output:
      #...
    plugins:
      - name: my-plugin
        artifacts:
          - type: other
            url: https://my-domain.tld/my-other-file.ext
            sha512sum: 589...ab4
            fileName: name-of-file.ext
  #...

See Build schema reference