Configuración del conector de tipo Sink de Kafka de EventBridge para MSK Connect - Transmisión administrada de Amazon para Apache Kafka
Requisitos previosConfiguración de los recursos necesarios para MSK ConnectCreación del conectorEnvío de mensajes a Kafka

Configuración del conector de tipo Sink de Kafka de EventBridge para MSK Connect

Este tema muestra cómo configurar el conector Kafka de tipo Sink de EventBridge para MSK Connect. Este conector permite enviar eventos desde el clúster de MSK hacia los buses de eventos de EventBridge. Este tema describe el proceso para crear los recursos necesarios y configurar el conector con el fin de habilitar un flujo de datos continuo entre Kafka y EventBridge.

Requisitos previos

Antes de implementar el conector, asegúrese de contar con los siguientes recursos:

  • Clúster de Amazon MSK: un clúster de MSK activo para producir y consumir mensajes de Kafka.

  • Bus de eventos de Amazon EventBridge: un bus de eventos de EventBridge para recibir eventos desde los temas de Kafka.

  • Roles de IAM: cree roles de IAM con los permisos necesarios para MSK Connect y para el conector de EventBridge.

  • Acceso a Internet público desde MSK Connect o un punto de conexión de interfaz de VPC para EventBridge, creado en la VPC y la subred del clúster de MSK. Esto permite evitar el tránsito por Internet público sin necesidad de utilizar puertas de enlace NAT.

  • Una máquina cliente, como una instancia de Amazon EC2 o AWS CloudShell, para crear temas y enviar registros a Kafka.

Configuración de los recursos necesarios para MSK Connect

Crea un rol de IAM para el conector y, posteriormente, crea el conector. También crea una regla de EventBridge para filtrar los eventos de Kafka enviados al bus de eventos de EventBridge.

Rol de IAM para el conector

El rol de IAM que asocie al conector debe contar con el permiso PutEvents para permitir el envío de eventos a EventBridge. El siguiente ejemplo de política de IAM le concede el permiso para enviar eventos a un bus de eventos denominado example-event-bus. Asegúrese de reemplazar el ARN del recurso en el siguiente ejemplo por el ARN del bus de eventos.

JSON
{
  "Version":"2012-10-17",		 	 	 
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "events:PutEvents"
      ],
      "Resource": "arn:aws:events:us-east-1:123456789012:event-bus/example-event-bus"
    }
  ]
}

Además, debe asegurarse de que el rol de IAM del conector contenga la siguiente política de confianza.

JSON
{
  "Version":"2012-10-17",		 	 	 
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "kafkaconnect.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

Una regla de EventBridge para eventos entrantes

Crea reglas que coinciden con los eventos entrantes en función de criterios de datos de eventos, conocidos como patrones de eventos. Con un patrón de eventos, puede definir los criterios para filtrar eventos entrantes y determinar qué eventos deben activar una regla específica y, posteriormente, ser dirigidos a un destino designado. El siguiente ejemplo de patrón de eventos coincide con eventos de Kafka enviados al bus de eventos de EventBridge.

{
  "detail": {
    "topic": ["msk-eventbridge-tutorial"]
  }
}

A continuación se muestra un ejemplo de un evento enviado desde Kafka a EventBridge mediante el conector de tipo Sink de Kafka.

{
  "version": "0",
  "id": "dbc1c73a-c51d-0c0e-ca61-ab9278974c57",
  "account": "123456789012",
  "time": "2025-03-26T10:15:00Z",
  "region": "us-east-1",
  "detail-type": "msk-eventbridge-tutorial",
  "source": "kafka-connect.msk-eventbridge-tutorial",
  "resources": [],
  "detail": {
    "topic": "msk-eventbridge-tutorial",
    "partition": 0,
    "offset": 0,
    "timestamp": 1742984100000,
    "timestampType": "CreateTime",
    "headers": [],
    "key": "order-1",
    "value": {
      "orderItems": [
        "item-1",
        "item-2"
      ],
      "orderCreatedTime": "Wed Mar 26 10:15:00 UTC 2025"
    }
  }
}

En la consola de EventBridge, cree una regla en el bus de eventos mediante este patrón de ejemplo y especifique un destino, como un grupo de Registros de CloudWatch. La consola de EventBridge configurará automáticamente la política de acceso necesaria para el grupo de Registros de CloudWatch.

Creación del conector

En la siguiente sección, crea e implementa el conector de tipo Sink de Kafka de EventBridge mediante la Consola de administración de AWS.

Paso 1: Descarga del conector

Descargue el archivo JAR más reciente del conector de tipo Sink de EventBridge desde la página de versiones de GitHub del conector de Kafka de EventBridge. Por ejemplo, para descargar la versión v1.4.1, seleccione el enlace del archivo JAR, kafka-eventbridge-sink-with-dependencies.jar, para descargar el conector. A continuación, guarde el archivo en una ubicación de su preferencia en la máquina.

Paso 2: creación de un bucket de Amazon S3

  1. Para almacenar el archivo JAR en Amazon S3 para su uso con MSK Connect, abra la Consola de administración de AWS y luego seleccione Amazon S3.

  2. En la consola de Amazon S3, seleccione Crear bucket e introduzca un nombre de bucket único. Por ejemplo, amzn-s3-demo-bucket1-eb-connector.

  3. Seleccione una región adecuada para el bucket de Amazon S3. Asegúrese de que coincida con la región en la que está implementado el clúster de MSK.

  4. En Configuración del bucket, mantenga las selecciones predeterminadas o ajústelas según sea necesario.

  5. Elija Crear bucket

  6. Cargue el archivo JAR en el bucket de Amazon S3.

Paso 3: Creación de un complemento en MSK Connect

  1. Abra la Consola de administración de AWS y luego vaya a MSK Connect.

  2. En el panel de navegación izquierdo, seleccione Complementos personalizados.

  3. Seleccione Crear complemento y, a continuación, introduzca un Nombre del complemento. Por ejemplo, eventbridge-sink-plugin.

  4. En Ubicación del complemento personalizado, pegue la URL del objeto de S3.

  5. Agregue una descripción opcional para el complemento.

  6. Seleccione Crear complemento.

Después de crear el complemento, puede utilizarlo para configurar e implementar el conector de tipo Sink de Kafka de EventBridge en MSK Connect.

Paso 4: creación del conector

Antes de crear el conector, se recomienda crear el tema de Kafka requerido para evitar errores del conector. Para crear el tema, utilice la máquina cliente.

  1. En el panel de navegación izquierdo de la consola de MSK, seleccione Conectores y, a continuación, elija Crear conector.

  2. En la lista de complementos, seleccione eventbridge-sink-plugin y luego elija Siguiente.

  3. En nombre del conector, introduzca EventBridgeSink.

  4. En la lista de clústeres, seleccione el clúster de MSK.

  5. Copie la siguiente configuración del conector y péguela en el campo Configuración del conector

    Reemplace los marcadores de posición en la siguiente configuración según sea necesario.

    • Elimine aws.eventbridge.endpoint.uri si el clúster de MSK tiene acceso público a Internet.

    • Si utiliza PrivateLink para conectarse de forma segura desde MSK a EventBridge, reemplace la parte del DNS posterior a https:// por el nombre DNS privado correcto del punto de conexión de la interfaz de VPC de EventBridge (opcional) que creó anteriormente.

    • Reemplace el ARN del bus de eventos de EventBridge en la siguiente configuración por el ARN del bus de eventos.

    • Actualice cualquier valor específico de la región.

    {
  "connector.class": "software.amazon.event.kafkaconnector.EventBridgeSinkConnector",
  "aws.eventbridge.connector.id": "msk-eventbridge-tutorial",
  "topics": "msk-eventbridge-tutorial",
  "tasks.max": "1",
  "aws.eventbridge.endpoint.uri": "https://events.us-east-1.amazonaws.com",
  "aws.eventbridge.eventbus.arn": "arn:aws:events:us-east-1:123456789012:event-bus/example-event-bus",
  "value.converter.schemas.enable": "false",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "aws.eventbridge.region": "us-east-1",
  "auto.offset.reset": "earliest",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter"
}

    Para obtener más información sobre la configuración de conectores, consulte eventbridge-kafka-connector.

    Si es necesario, modifique la configuración de los procesos de trabajo y del escalado automático. También se recomienda utilizar la versión más reciente disponible (recomendada) de Apache Kafka Connect desde la lista desplegable. En Permisos de acceso, utilice el rol creado anteriormente. Asimismo, se recomienda habilitar el registro en CloudWatch para mejorar la observabilidad y facilitar la solución de problemas. Ajuste las demás opciones discrecionales, como las etiquetas, según sus necesidades. A continuación, implemente el conector y espere a que el estado pase a En ejecución.

Envío de mensajes a Kafka

Puede configurar codificaciones de mensajes, como Apache Avro y JSON, mediante la especificación de convertidores con el value.converter y, de forma opcional, mediante configuraciones de key.converter disponibles en Kafka Connect.

El connector example de este tema está configurado para trabajar con mensajes codificados en JSON, tal como se indica mediante el uso de org.apache.kafka.connect.json.JsonConverter para value converter. Cuando el conector se encuentre en estado En ejecución, envíe registros al tema msk-eventbridge-tutorial de Kafka desde la máquina cliente.