Interroger les sujets de Kafka en mode file d'attente

Le mode file d'attente est un mode de consommation qui utilise des groupes de partage, qui est devenu disponible dans Apache Kafka 4.0 (le Kafka Queue Poller nécessite la version 4.2 ou ultérieure). Avec le mode File d'attente, vous pouvez traiter les messages thématiques de Kafka de manière non ordonnée et à haut débit à l'aide de groupes de partage. Contrairement aux groupes de consommateurs Kafka traditionnels qui attribuent des partitions exclusivement aux consommateurs, les groupes de partage permettent à plusieurs consommateurs de lire simultanément à partir des mêmes partitions. Cette méthode est similaire à celle utilisée par Amazon SQS pour traiter les messages provenant d'une file d'attente.

Lorsque vous créez un mappage de source d'événements pour un environnement autogéré Apache Kafka cluster, vous pouvez choisir entre deux modes de consommation :

• Diffusez (par défaut). Utilise les groupes de consommateurs Kafka avec un ordre par partition. Lambda attribue chaque partition à un seul consommateur.

• File d'attente. Utilise des groupes de partage, qui sont devenus disponibles dans Apache Kafka 4.0, pour un traitement à haut débit non ordonné (le Kafka Queue Poller nécessite la version 4.2 ou ultérieure). Avec les groupes de partage, plusieurs consommateurs peuvent lire simultanément à partir des mêmes partitions, de la même manière qu'Amazon SQS traite les messages d'une file d'attente.

Si vous ne spécifiez aucun mode de consommation, le mappage des sources d'événements utilise Stream. Cette page couvre le mode de consommation des files d'attente. Pour des informations générales sur l'utilisation de Lambda avec l'autogestion Apache Kafka, voir Utilisation de Lambda avec Apache Kafka autogéré.

En mode File d'attente, Lambda utilise le Kafka Share Consumer pour interroger les messages du sujet. Chaque sondeur d'événements traite les messages sur toutes les partitions sans ordre au niveau des partitions. Cela augmente la simultanéité pour chaque sondeur.

Configuration requise

Pour utiliser le mode de consommation des files d'attente, votre cluster Kafka doit répondre aux exigences suivantes :

• Apache Kafka version 4.2 ou ultérieure avec les groupes de partage activés. Kafka permet share.version par défaut, et vous pouvez le gérer via kafka-features.sh.

• Self-managed Apache Kafka Clusters (SMK) uniquement. Streaming géré par Amazon pour Apache Kafka (Amazon MSK) ne prend pas actuellement en charge le mode File d'attente. AWS ajoute le support MSK une fois que MSK propose Kafka 4.2 ou une version ultérieure avec les groupes de partage activés.

Comparaison des flux et des files d'attente

Fonctionnalité	Stream (par défaut)	File d’attente
Commander	Per-partition commande garantie	Aucune garantie de commande
Modèle de consommation	Groupe de consommateurs (cession exclusive de partage)	Groupe de partage (accès simultané aux partitions)
Débit par sondeur d'événements	Jusqu'à 5 MB/s	Jusqu'à 5 MB/s
Invocations simultanées par participant à l'événement	Jusqu’à 5	Jusqu’à 10
Modes de dimensionnement	On-demand et approvisionné	On-demand et approvisionné
Partage de l'EPU	Peut partager des EPU avec d'autres mappages de sources d'événements Kafka Streams	Isolé. Les mappages de sources d'événements de file d'attente ne sont partagés qu'avec d'autres mappages de sources d'événements de file d'attente
Cas d’utilisation	Traitement ordonné des événements	High-throughput charges de travail non ordonnées

Création d'un mappage de file

Pour créer un mappage de sources d'événements en mode file d'attente, définissez ConsumptionMode domaine de SelfManagedKafkaEventSourceConfig pour Queue. Dans les exemples suivants, remplacez my-function, les points de terminaison du courtier, et my-queue-topic avec vos propres valeurs.

AWS INTERFACE DE LIGNE DE COMMANDE (CLI)

Les suivants AWS La commande CLI crée un mappage de source d'événements en mode file d'attente. Remplacer my-function avec le nom de votre fonction, le broker se termine par vos courtiers, et my-queue-topic avec votre sujet.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 1, "MaximumPollers": 10}'

AWS CloudFormation

Les suivants AWS CloudFormation Le modèle crée un mappage de source d'événements en mode file d'attente. Remplacez les points de terminaison du courtier et my-queue-topic avec vos propres valeurs.

MyKafkaQueueESM:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    FunctionName: !Ref MyFunction
    SelfManagedEventSource:
      Endpoints:
        KafkaBootstrapServers:
          - broker1.us-west-2.example.com:9092
          - broker2.us-west-2.example.com:9092
    Topics:
      - my-queue-topic
    BatchSize: 100
    SelfManagedKafkaEventSourceConfig:
      ConsumptionMode: Queue
    ProvisionedPollerConfig:
      MinimumPollers: 1
      MaximumPollers: 10

Paramètres de configuration

Le tableau suivant répertorie les paramètres applicables aux mappages de sources d'événements en mode file d'attente.

Paramètre	Description	Plage/Par défaut
BatchSize	Nombre maximum d'enregistrements par lot envoyés à la fonction	1 à 10 000. Par défaut : 100
Enabled	Si le mappage des sources d'événements est actif	true ou false. Valeur par défaut : true
FilterCriteria	Modèles de filtrage des événements pour traiter uniquement les enregistrements correspondants	Jusqu'à 5 modèles de filtre
MaximumRetryAttempts	Nombre maximum de tentatives pour les enregistrements ayant échoué	-1 à 10 000. Par défaut : 0 (en mode file d'attente, le broker Kafka effectue de nouvelles tentatives lorsqu'aucune destination en cas d'échec n'est configurée)
BisectBatchOnFunctionError	Diviser le lot en deux en cas d'erreur de fonction	true ou false. Valeur par défaut : false
ProvisionedPollerConfig.MinimumPollers	Nombre minimal de participants aux sondages d'événements (mode provisionné)	1 à 200. Valeur par défaut : 1
ProvisionedPollerConfig.MaximumPollers	Nombre maximum de personnes interrogées sur les événements (mode provisionné)	1 à 2 000. Par défaut : 200
ProvisionedPollerConfig.PollerGroupName	Nom du groupe pour le partage EPU (mode provisionné, mappages de sources d'événements en file d'attente uniquement)	String. Jusqu'à 100 mappages de sources d'événements par groupe
SelfManagedKafkaEventSourceConfig.ConsumptionMode	Mode de consommation pour la source de l'événement	Doit être Queue pour le mode file d'attente
SchemaRegistryConfig	Registre de schémas pour la validation des événements	Consultez Utilisation de registres de schémas avec les sources d’événements Kafka dans Lambda

Important

Le MaximumBatchingWindowInSeconds le paramètre est non pris en charge pour les mappages de sources d'événements en mode file d'attente. Le consommateur de Kafka Share utilise une reconnaissance explicite. Tous les enregistrements d'un lot doivent faire l'objet d'un accusé de réception explicite avant que le consommateur puisse en récupérer de nouveaux. Cela n'est pas compatible avec une fenêtre de traitement par lots basée sur le temps. Si vous spécifiez MaximumBatchingWindowInSeconds lors de la création d'un mappage de source d'événements en mode file d'attente, l'API renvoie une erreur de validation.

Fonctionnalités de mappage des sources d'événements prises en charge

Le tableau suivant répertorie les fonctionnalités de mappage des sources d'événements Lambda disponibles avec le mode Queue.

Fonctionnalité	Pris en charge	Remarques
Mode alloué	Oui	Configurez MinimumPollers et MaximumPollers
On-demand mode	Oui	Mode de mise à l'échelle par défaut. Lambda balance automatiquement les sondeurs jusqu'à 5 MB/s ou 50 appels simultanés, selon la première valeur atteinte. Aucune configuration supplémentaire n'est requise.
Filtrage des événements	Oui	Jusqu'à 5 modèles de filtre par mappage de source d'événement
Registre de schémas	Oui	Avro et Protobuf avec Glue Schema Registry ou Confluent Schema Registry. Le filtrage du registre des schémas n'est disponible qu'en mode provisionné.
Low-latency En traitement	Oui	Disponible en mode provisionné
Réessayez les configurations	Oui	MaximumRetryAttempts, BisectBatchOnFunctionError
On-failure destination (thème de Kafka)	Oui	Envoyer les enregistrements ayant échoué à un sujet Kafka
On-failure destination (SNS/SQS)	Oui	Envoyer les enregistrements ayant échoué à SNS ou SQS
Enregistrement du mappage des sources d'événements	Oui	CloudWatch Journaux pour les diagnostics de mappage des sources d'événements
Groupement de mappages de sources d'événements (PollerGroupName)	Oui	Les mappages de sources d'événements en mode file d'attente peuvent partager des EPU avec d'autres mappages de sources d'événements en mode file d'attente uniquement (mode provisionné)
Fenêtre de traitement par lots	Non	Non pris en charge en raison du comportement du mode de blocage Share Consumer
Position de départ	Non	StartingPosition et StartingPositionTimestamp ne sont pas prises en charge. Le mode file d'attente lit toujours à partir de la position validée du groupe de partage.

Modes de dimensionnement

Les mappages de sources d'événements en mode file d'attente prennent en charge à la fois les modes de dimensionnement à la demande (par défaut) et provisionnés.

On-demand mode (par défaut)

En mode à la demande, Lambda gère automatiquement les sondeurs d'événements pour votre mappage des sources d'événements en mode file d'attente.

On-demand le mode permet d'effectuer jusqu'à 5 sondages d'événements par mappage de source d'événements. Il atteint son plafond à 5 MB/s débit global ou 50 appels simultanés (5 interrogateurs multipliés par 10). Cela se produit quelle que soit la limite atteinte en premier. Pour dépasser ces limites, utilisez le mode provisionné. On-demand le mode ne nécessite aucune configuration supplémentaire au-delà des paramètres de mappage des sources d'événements standard.

Le tableau suivant répertorie les limites du mode à la demande.

Limite	Value
Débit par sondeur d'événements	Jusqu'à 1 MB/s
Invocations simultanées par participant à l'événement	Jusqu’à 10
Nombre maximum de participants à l'événement	5
Débit agrégé maximal	Jusqu'à 5 MB/s (5 personnes interrogées fois 1) MB/s)
Simultanéité agrégée maximale	Jusqu'à 50 invocations simultanées (5 sondeurs multipliés par 10)

Exemple — Mappage des sources d'événements en mode file d'attente en mode à la demande (AWS CLI)

Ce qui suit AWS La commande CLI crée un mappage de source d'événements en mode file d'attente en mode à la demande. Remplacer my-function, les points de terminaison du courtier, my-queue-topic, les ID de sous-réseau et l'ID du groupe de sécurité avec vos propres valeurs.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

Mode alloué

Pour les charges de travail soumises à des exigences de performances strictes ou à des modèles de trafic élevés, utilisez le mode provisionné. Le mode provisionné vous permet de contrôler directement le nombre minimum et maximum de sondeurs d'événements dédiés à votre mappage des sources d'événements en mode file d'attente.

Chaque sondeur d'événements en mode file d'attente en mode provisionné prend en charge jusqu'à 5 MB/s de débit et jusqu'à 10 appels Lambda simultanés.

Le tableau suivant répertorie les limites du mode provisionné.

Limite	Value
Débit par sondeur d'événements	Jusqu'à 5 MB/s
Invocations simultanées par participant à l'événement	Jusqu’à 10
Nombre minimum de participants à l'événement (MinimumPollers)	1 à 200. Valeur par défaut : 1
Nombre maximum de participants à l'événement (MaximumPollers)	1 à 2 000. Par défaut : 200
Mappages des sources d'événements par groupe de sondeurs	Jusqu'à 100
Nombre maximum total de personnes interrogées par groupe de sondeurs	2 000
Débit par EPU (agrégé)	Jusqu'à 5 MB/s par poller ; l'agrégat dépend de la densité de remplissage du poller

Le mode provisionné est facturé en fonction de l'utilisation de l'Event Poller Unit (EPU). Pour plus d'informations sur la tarification, consultez la page AWS Lambda Pricing (Tarification).

Exemple — Mappage de sources d'événements en mode file d'attente avec le mode provisionné (AWS CLI)

Ce qui suit AWS La commande CLI crée un mappage de source d'événements en mode file d'attente en mode provisionné.

PROMPT> aws lambda create-event-source-mapping \
  --function-name my-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["broker1.us-west-2.example.com:9092","broker2.us-west-2.example.com:9092"]}}' \
  --topics my-queue-topic \
  --batch-size 100 \
  --self-managed-kafka-event-source-config '{"ConsumptionMode": "Queue"}' \
  --provisioned-poller-config '{"MinimumPollers": 5, "MaximumPollers": 100}' \
  --source-access-configurations '[{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef0"},{"Type":"VPC_SUBNET","URI":"subnet-0123456789abcdef1"},{"Type":"VPC_SECURITY_GROUP","URI":"sg-0123456789abcdef0"}]'

Remplacez les espaces réservés suivants par vos propres valeurs :

• subnet-0123456789abcdef0 et subnet-0123456789abcdef1: vos identifiants de sous-réseau.

• sg-0123456789abcdef0: l'identifiant de votre groupe de sécurité.

• MinimumPollers et MaximumPollers: paramétré en fonction de vos besoins en matière de débit.

Note

Pour passer du mode provisionné au mode à la demande, passez un champ vide ProvisionedPollerConfig:

PROMPT> aws lambda update-event-source-mapping \
  --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
  --provisioned-poller-config '{}'

Sélection du mode de dimensionnement

Considération	On-demand	Alloué
Configuration	Aucune configuration supplémentaire	Configurez MinimumPollers et MaximumPollers
Coût	Pas de frais EPU supplémentaires	Frais EPU basés sur l'utilisation par heure
Idéal pour	Low-concurrency charges de travail	Exigences de latence strictes, trafic élevé
Groupement de mappages de sources d'événements (PollerGroupName)	Non pris en charge	Pris en charge. Partagez des EPU entre les mappages de sources d'événements de la file d'attente
Low-latency En traitement	Non disponible	Available
Filtrage du registre des schémas (Avro/Protobuf)	Non disponible	Available

EPU partageant l'isolation

En mode provisionné, vous pouvez regrouper plusieurs mappages de sources d'événements au sein du même VPC afin de partager la capacité de l'Event Poller Unit (EPU) en leur attribuant la même capacité PollerGroupName. Le regroupement de mappages de sources d'événements permet de regrouper des mappages de sources d'événements à faible débit sur des EPU partagés au lieu de nécessiter chacun une capacité dédiée, ce qui réduit les coûts du mode provisionné. Pour plus de détails, consultez Optimisation des coûts pour le mode provisionné.

Les mappages de sources d'événements en mode file d'attente ne partagent les EPU qu'avec d'autres mappages de sources d'événements en mode file d'attente. UNE PollerGroupName utilisé par un mappage de source d'événements en mode Queue ne peut pas être partagé avec un mappage de source d'événements Kafka Streams, et vice versa. Si vous configurez un PollerGroupName déjà utilisé par l'autre mode de consommation, CreateEventSourceMapping renvoie une erreur de validation. Pour utiliser le regroupement de mappages de sources d'événements avec le mode File d'attente, créez un PollerGroupName dédié à vos mappages de sources d'événements de file d'attente.

Exemple d’évènement

Lorsque Lambda appelle votre fonction avec une source d'événements en mode Queue, la charge utile de l'événement est identique à celle d'un événement Kafka standard. Chaque enregistrement inclut le sujet, la partition, le décalage, l'horodatage, ainsi que la clé et la valeur codées en base64 :

{
  "eventSource": "SelfManagedKafka",
  "bootstrapServers": "broker1.example.com:9092,broker2.example.com:9092",
  "records": {
    "my-queue-topic-0": [
      {
        "topic": "my-queue-topic",
        "partition": 0,
        "offset": 15,
        "timestamp": 1545084650987,
        "timestampType": "CREATE_TIME",
        "key": "abcDEFghiJKLmnoPQRstuVWXyz1234==",
        "value": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
        "headers": [
          {
            "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101]
          }
        ]
      }
    ],
    "my-queue-topic-3": [
      {
        "topic": "my-queue-topic",
        "partition": 3,
        "offset": 42,
        "timestamp": 1545084651000,
        "timestampType": "CREATE_TIME",
        "key": "YW5vdGhlcktleQ==",
        "value": "VGhpcyBpcyBhbm90aGVyIG1lc3NhZ2U=",
        "headers": []
      }
    ]
  }
}

Idempotence requise

Les mappages de sources d'événements Lambda traitent chaque événement au moins une fois, ce qui peut entraîner un traitement dupliqué des enregistrements. Le mode Queue ne garantissant pas l'ordre, le traitement des doublons ou des erreurs de traitement peut produire des résultats incorrects, tels qu'un double comptage ou un état incohérent, si le code de votre fonction n'est pas idempotent. Rendez votre code de fonction idempotent.

Limitations

• Self-managed Clusters Kafka uniquement. Amazon MSK n'est pas pris en charge au lancement.

• Apache Kafka 4.2 ou version ultérieure requise. Le cluster doit prendre en charge le protocole Share Consumer avec les groupes de partage activés.

• La fenêtre de traitement par lots n'est pas prise en charge. Le MaximumBatchingWindowInSeconds Le paramètre ne peut pas être utilisé avec les mappages de sources d'événements Kafka Queue.

• Position de départ non prise en charge. StartingPosition et StartingPositionTimestamp ne peut pas être utilisé avec les mappages de sources d'événements en mode file d'attente. Le groupe d'actions consomme à partir de sa position de compensation engagée.

• Isolation EPU. Les mappages de sources d'événements en mode file d'attente ne peuvent pas partager les EPU avec les mappages de sources d'événements de Kafka Streams. Utiliser séparément PollerGroupName valeurs.