Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés. # Descripción de MSK Connect MSK Connect es una característica de Amazon MSK que facilita a los desarrolladores la transmisión de datos hacia y desde sus clústeres de Apache Kafka. MSK Connect utiliza versiones de Kafka Connect 2.7.1 o 3.7.x, que son marcos de trabajo de código abierto para conectar clústeres de Apache Kafka con sistemas externos, como bases de datos, índices de búsqueda y sistemas de archivos. Con MSK Connect, puede implementar conectores totalmente gestionados diseñados para Kafka Connect que mueven o extraen datos de almacenes de datos populares, como Amazon S3 y Amazon OpenSearch Service. Puede implementar conectores desarrollados por terceros, como Debezium, para transmitir los registros de cambios de las bases de datos a un clúster de Apache Kafka, o implementar un conector existente sin cambios de código. Los conectores se escalan automáticamente para adaptarse a los cambios de carga y solo pagará por los recursos que utilice. Utilice los conectores de origen para importar datos de sistemas externos a sus temas. Con los conectores de recepción, puede exportar datos de sus temas a sistemas externos. MSK Connect admite conectores para cualquier clúster de Apache Kafka con conectividad a una Amazon VPC, ya sea un clúster de MSK o un clúster de Apache Kafka alojado de forma independiente. MSK Connect monitorea continuamente el buen estado general y el estado de entrega de los conectores, parchea y administra el hardware subyacente y escala automáticamente los conectores para adaptarlos a los cambios en el rendimiento. Para comenzar a utilizar MSK Connect, consulte [Introducción a MSK Connect](msk-connect-getting-started.md). Para obtener información sobre los AWS recursos que puede crear con MSK Connect, consulte [Descripción de los conectores](msk-connect-connectors.md)[Creación de complementos personalizados](msk-connect-plugins.md), y[Descripción de los procesos de trabajo de MSK Connect](msk-connect-workers.md). Para obtener información sobre la API de MSK Connect, consulte la [Referencia de la API de Amazon MSK Connect](https://docs.aws.amazon.com/MSKC/latest/mskc/Welcome.html). ## Beneficios de usar Amazon MSK Connect Apache Kafka es una de las plataformas de transmisión de código abierto más utilizadas para la ingesta y el procesamiento de flujos de datos en tiempo real. Con Apache Kafka, puede desacoplar y escalar de forma independiente las aplicaciones que producen y consumen datos. Kafka Connect es un componente importante para la creación y la ejecución de aplicaciones de transmisión con Apache Kafka. Kafka Connect proporciona una forma estandarizada de trasladar datos entre Kafka y los sistemas externos. Kafka Connect es altamente escalable y puede administrar grandes volúmenes de datos. Kafka Connect proporciona un potente conjunto de operaciones y herramientas de API para configurar, implementar y supervisar conectores que trasladan datos entre los temas de Kafka y sistemas externos. Con estas herramientas, puede personalizar y ampliar la funcionalidad de Kafka Connect para satisfacer las necesidades específicas de su aplicación de transmisión. Es posible que encuentre dificultades cuando utilice clústeres de Apache Kafka Connect por sí solos o cuando intente migrar aplicaciones de código abierto de Apache Kafka Connect a AWS. Estos desafíos incluyen el tiempo necesario para configurar la infraestructura e implementar las aplicaciones, los obstáculos de ingeniería al configurar los clústeres de Apache Kafka Connect autoadministrados y la sobrecarga operativa administrativa. Para hacer frente a estos desafíos, recomendamos que utilice Amazon Managed Streaming para Apache Kafka Connect (Amazon MSK Connect) para migrar sus aplicaciones de código abierto Apache Kafka Connect a AWS. Amazon MSK Connect simplifica el uso de Kafka Connect para transmitir datos entre los clústeres de Apache Kafka y sistemas externos, como bases de datos, índices de búsqueda y sistemas de archivos. Estas son algunas de las ventajas de migrar a Amazon MSK Connect: + **Eliminación de la sobrecarga operativa**: Amazon MSK Connect elimina la carga operativa asociada a la aplicación de parches, el aprovisionamiento y el escalado de los clústeres de Apache Kafka Connect. Amazon MSK Connect supervisa de forma continua el estado de los clústeres de Connect y automatiza la aplicación de parches y las actualizaciones de versiones sin interrumpir las cargas de trabajo. + **Reinicio automático de las tareas de Connect**: Amazon MSK Connect puede recuperar automáticamente las tareas fallidas para reducir las interrupciones en la producción. Los errores en las tareas pueden ser el resultado de errores temporales, como superar el límite de conexiones TCP en Kafka o el reequilibrio de tareas cuando se incorporan nuevos procesos de trabajo al grupo de consumidores de conectores de recepción. + **Escalado vertical y horizontal automático**: Amazon MSK Connect permite que el conector de aplicaciones se escale automáticamente para que sea compatible con mayores rendimientos. Amazon MSK Connect administra el escalado por usted. Solo necesita especificar el número de procesos de trabajo en el grupo de escalado automático y los umbrales de utilización. Puede usar la operación de la `UpdateConnector` API Amazon MSK Connect para escalar verticalmente hacia arriba o hacia abajo la v CPUs entre 1 y 8 v CPUs para admitir un rendimiento variable. + **Conectividad de red privada**: Amazon MSK Connect se conecta de forma privada a los sistemas de origen AWS PrivateLink y destino mediante nombres DNS privados. # Introducción a MSK Connect Este es un step-by-step tutorial que utiliza el Consola de administración de AWS para crear un clúster de MSK y un conector receptor que envía los datos del clúster a un bucket de S3. **Topics** + [Configuración de los recursos necesarios para MSK Connect](mkc-tutorial-setup.md) + [Creación de un complemento personalizado](mkc-create-plugin.md) + [Creación de un equipo cliente y un tema de Apache Kafka](mkc-create-topic.md) + [Crea conector](mkc-create-connector.md) + [Envío de datos al clúster de MSK](mkc-send-data.md) # Configuración de los recursos necesarios para MSK Connect En este paso, creará los siguientes recursos que necesitará para este escenario de introducción: + Un bucket de Amazon S3 que actúa como destino y recibe los datos del conector. + Un clúster de MSK al que enviará los datos. A continuación, el conector leerá los datos de este clúster y los enviará al bucket de S3 de destino. + Una política de IAM que contiene los permisos para escribir en el bucket de destino de S3. + Un rol de IAM que permite al conector escribir en el bucket de S3 de destino. Agregará la política de IAM que cree a este rol. + Un punto de conexión de VPC de Amazon VPC para poder enviar datos desde la Amazon VPC que tiene el clúster y el conector a Amazon S3. **Creación del bucket de S3** 1. Inicie sesión en la consola de Amazon S3 Consola de administración de AWS y ábrala en [https://console.aws.amazon.com/s3/](https://console.aws.amazon.com/s3/). 1. Elija **Crear bucket**. 1. Para el nombre del bucket, introduzca un nombre descriptivo, como `amzn-s3-demo-bucket-mkc-tutorial`. 1. Desplácese hacia abajo y seleccione **Crear bucket**. 1. En la lista de buckets, elija el bucket que acaba de crear. 1. Elija **Crear carpeta**. 1. Introduzca `tutorial` para el nombre de la carpeta, desplácese hacia abajo y elija **Crear carpeta**. **Creación del clúster** 1. ¿Abrir la consola Amazon MSK en [https://console.aws.amazon.com/msk/casa? region=us-east-1\$1/home/](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/). 1. En el panel izquierdo, en **Clústeres de MSK**, seleccione **Clústeres**. 1. Elija **Create cluster**. 1. En **Método de creación**, elija **Creación personalizada**. 1. Para el nombre del clúster, ingrese **mkc-tutorial-cluster**. 1. En **Tipo de clúster**, elija **Aprovisionado**. 1. Elija **Siguiente**. 1. En **Redes**, elija una Amazon VPC. A continuación, seleccione las zonas de disponibilidad y las subredes que desee utilizar. Recuerde IDs la VPC y las subredes de Amazon que seleccionó porque las necesitará más adelante en este tutorial. 1. Elija **Siguiente**. 1. En **Métodos de control de acceso**, asegúrese de seleccionar solo **Acceso no autenticado**. 1. En **Cifrado**, asegúrese de seleccionar solo **Texto sin formato**. 1. Continúe con el asistente y, a continuación, seleccione **Crear clúster**. Esto le lleva a la página Detalles del clúster. En esa página, en **Grupos de seguridad aplicados**, busque el ID del grupo de seguridad. Recuerde ese ID, ya que lo necesitará más tarde en este tutorial. **Para crear una política de IAM con permisos para escribir en el bucket de S3** 1. Abra la consola de IAM en [https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/). 1. En el panel de navegación, seleccione **Políticas**. 1. Elija **Crear política**. 1. En **Editor de políticas**, elija **JSON** y, a continuación, reemplace el JSON en la ventana del editor por el siguiente JSON. En el siguiente ejemplo, ** sustitúyalo por el nombre del bucket de S3. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "AllowListBucket", "Effect": "Allow", "Action": [ "s3:ListBucket", "s3:GetBucketLocation" ], "Resource": "arn:aws:s3:::" }, { "Sid": "AllowObjectActions", "Effect": "Allow", "Action": [ "s3:PutObject", "s3:GetObject", "s3:DeleteObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts", "s3:ListBucketMultipartUploads" ], "Resource": "arn:aws:s3:::/*" } ] } ``` ------ Para obtener instrucciones sobre cómo escribir políticas seguras, consulte [Control de acceso de IAM](iam-access-control.md). 1. Elija **Siguiente**. 1. En la página **Revisar y crear**, haga lo siguiente: 1. En **Nombre de la política**, introduzca un nombre descriptivo, como **mkc-tutorial-policy**. 1. En **Permisos definidos en esta política**, revise y and/or edite los permisos definidos en su política. 1. (Opcional) Para ayudar a identificar, organizar o buscar la política, elija **Agregar nueva etiqueta** para agregar etiquetas como pares clave-valor. Por ejemplo, agregue una etiqueta a la política con el par clave-valor de **Environment** y **Test**. Para obtener más información sobre el uso de etiquetas, consulte [Etiquetas para AWS Identity and Access Management recursos](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_tags.html) en la *Guía del usuario de IAM*. 1. Elija **Crear política**. **Creación del rol de IAM que pueda escribir en el bucket de destino** 1. En el panel de navegación de la consola de IAM, elija **Roles** y, a continuación, seleccione **Crear rol**. 1. En la página **Seleccionar entidad de confianza**, haga lo siguiente: 1. En **Tipo de entidad de confianza**, elija **Servicio de AWS**. 1. En **Servicio o caso de uso**, elija **S3**. 1. En **Caso de uso**, elija **S3**. 1. Elija **Siguiente**. 1. En la página **Agregar permisos**, haga lo siguiente: 1. En el cuadro de búsqueda de **Políticas de permisos**, introduzca el nombre de la política que creó previamente para este tutorial. Por ejemplo, **mkc-tutorial-policy**. A continuación, marque la casilla situada a la izquierda del nombre de la política. 1. (Opcional) Configure un [límite de permisos](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_boundaries.html). Se trata de una característica avanzada que está disponible para los roles de servicio, pero no para los roles vinculados a servicios. Para obtener información sobre cómo establecer un límite de permisos, consulte [Creación de roles y asociación de políticas (consola)](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_job-functions_create-policies.html) en la *Guía del usuario de IAM*. 1. Elija **Siguiente**. 1. En la página **Nombrar, revisar y crear**, haga lo siguiente: 1. En **Nombre del rol**, introduzca un nombre descriptivo, como **mkc-tutorial-role**. **importante** Cuando asigne un nombre a un rol, tenga en cuenta lo siguiente: Los nombres de los roles deben ser únicos dentro de tu Cuenta de AWS perfil y no se pueden hacer únicos por mayúsculas y minúsculas. Por ejemplo, no puedes crear roles denominados tanto **PRODROLE** como **prodrole**. Cuando se utiliza un nombre de rol en una política o como parte de un ARN, el nombre de rol distingue entre mayúsculas y minúsculas, sin embargo, cuando un nombre de rol les aparece a los clientes en la consola, como por ejemplo durante el proceso de inicio de sesión, el nombre de rol no distingue entre mayúsculas y minúsculas. Dado que otras entidades podrían hacer referencia al rol, no es posible editar el nombre del rol una vez creado. 1. (Opcional) **En Descripción**, ingrese una descripción para el rol. 1. (Opcional) Para editar los casos de uso y los permisos del rol, en las secciones **Paso 1: Seleccionar entidades de confianza** o **Paso 2: Agregar permisos**, elija **Editar**. 1. (Opcional) Para ayudar a identificar, organizar o buscar el rol, elija **Agregar nueva etiqueta** para agregar etiquetas como pares clave-valor. Por ejemplo, agregue una etiqueta al rol con el par clave-valor de **ProductManager** y **John**. Para obtener más información sobre el uso de etiquetas, consulte [Etiquetas para obtener AWS Identity and Access Management recursos](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_tags.html) en la *Guía del usuario de IAM*. 1. Revise el rol y, a continuación, elija **Crear rol**. **Autorización para que MSK Connect asuma el rol** 1. En la consola de IAM, en el panel izquierdo, en **Administración de acceso**, seleccione **Roles**. 1. Busque el `mkc-tutorial-role` y selecciónelo. 1. En el **Resumen** del rol, elija la pestaña **Relaciones de confianza**. 1. Elija **Editar relación de confianza**. 1. Reemplace la política existente por el siguiente JSON. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "kafkaconnect.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ 1. Elija **Actualizar política de confianza**. **Creación de un punto de conexión de VPC de Amazon VPC desde la VPC del clúster a Amazon S3** 1. Abra la consola de Amazon VPC en [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/). 1. En el panel izquierdo, elija **Puntos de conexión**. 1. Seleccione **Crear punto de conexión**. 1. En **Nombre del servicio**, elija el servicio **com.amazonaws.us-east-1.s3** y el tipo de **puerta de enlace**. 1. Elija la VPC del clúster y, a continuación, seleccione la casilla situada a la izquierda de la tabla de enrutamiento que está asociada a las subredes del clúster. 1. Seleccione **Crear punto de conexión**. **Paso siguiente** [Creación de un complemento personalizado](mkc-create-plugin.md) # Creación de un complemento personalizado Un complemento contiene el código que define la lógica del conector. En este paso, creará un complemento personalizado que contenga el código del conector de recepción de Amazon S3 de Lenses. En un paso posterior, cuando cree el conector de MSK, especifique que su código está en este complemento personalizado. Puede usar el mismo complemento para crear varios conectores de MSK con diferentes configuraciones. **Creación del complemento personalizado** 1. Descargue el [conector de S3](https://www.confluent.io/hub/confluentinc/kafka-connect-s3). 1. Cargue el archivo ZIP en un bucket de S3 al que tenga acceso. Para obtener información sobre cómo cargar archivos en Amazon S3, consulte [Carga de objetos](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html) en la guía del usuario de Amazon S3. 1. Abra la consola de Amazon MSK en [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/). 1. En el panel izquierdo, expanda **MSK Connect** y, a continuación, seleccione **Complementos personalizados**. 1. Seleccione **Crear complemento personalizado**. 1. Elija **Examinar S3**. 1. En la lista de buckets, busque el bucket en el que ha cargado el archivo ZIP, y elija ese bucket. 1. En la lista de objetos del bucket, seleccione el botón de opción situado a la izquierda del archivo ZIP y, a continuación, seleccione el botón denominado **Elegir**. 1. Ingrese `mkc-tutorial-plugin` para el nombre del complemento personalizado y, a continuación, seleccione **Crear complemento personalizado**. Es posible que tarde AWS unos minutos en terminar de crear el complemento personalizado. Cuando se complete el proceso de creación, verá el siguiente mensaje en un banner en la parte superior de la ventana del navegador. ``` Custom plugin mkc-tutorial-plugin was successfully created The custom plugin was created. You can now create a connector using this custom plugin. ``` **Paso siguiente** [Creación de un equipo cliente y un tema de Apache Kafka](mkc-create-topic.md) # Creación de un equipo cliente y un tema de Apache Kafka En este paso, cree una instancia de Amazon EC2 para utilizarla como instancia de cliente de Apache Kafka. A continuación, utilice esta instancia para crear un tema en el clúster. **Creación de un equipo cliente** 1. Abra la consola de Amazon EC2 en [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/). 1. Elija **iniciar instancias**. 1. Escriba un **nombre** para el equipo cliente, como **mkc-tutorial-client**. 1. Deje seleccionado **Amazon Linux 2 AMI (HVM): Kernel 5.10, tipo de volumen SSD** en **Tipo de imagen de máquina de Amazon (AMI)**. 1. Elija el tipo de instancia **t2.xlarge**. 1. En **Par de claves (inicio de sesión)**, seleccione **Crear un nuevo par de claves**. Introduzca **mkc-tutorial-key-pair** en **Nombre del par de claves** y, a continuación, seleccione **Descargar par de claves**. También puede utilizar un par de claves existente. 1. Seleccione **Iniciar instancia**. 1. Elija **View Instances (Ver instancias)**. A continuación, en la columna **Grupos de seguridad**, elija el grupo de seguridad asociado a la nueva instancia. Copie el ID del grupo de seguridad y guárdelo para más adelante. **Concesión de permiso al cliente recién creado para que envíe datos al clúster** 1. Abra la consola de Amazon VPC en [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/). 1. En el panel izquierdo, en **SEGURIDAD**, elija **Grupos de seguridad**. En la columna **ID de grupo de seguridad**, busque el grupo de seguridad del clúster. Guardó el ID de este grupo de seguridad cuando creó el clúster en [Configuración de los recursos necesarios para MSK Connect](mkc-tutorial-setup.md). Seleccione este grupo de seguridad seleccionando la casilla situada a la izquierda de su fila. Asegúrese de que no haya otros grupos de seguridad seleccionados simultáneamente. 1. En la parte inferior de la pantalla, elija la pestaña **Reglas de entrada**. 1. Elija **Editar reglas de entrada**. 1. En la parte inferior izquierda de la pantalla, elija **Añadir regla**. 1. En la nueva regla, elija **All traffic (Todo el tráfico)** en la columna **Type (Tipo)**. En el campo de la derecha de la columna **Origen**, escriba el ID del grupo de seguridad del equipo cliente. Este es el ID del grupo de seguridad que guardó después de crear el equipo cliente. 1. Seleccione **Guardar reglas**. Su clúster de MSK ahora aceptará todo el tráfico del cliente que creó en el procedimiento anterior. **Creación de un tema** 1. Abra la consola de Amazon EC2 en [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/). 1. En la tabla de instancias, elija `mkc-tutorial-client`. 1. Cerca de la parte superior de la pantalla, seleccione **Conectar** y siga las instrucciones para conectarse a la instancia. 1. Instale Java en la instancia del cliente ejecutando el siguiente comando: ``` sudo yum install java-1.8.0 ``` 1. Ejecute el siguiente comando para descargar Apache Kafka. ``` wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz ``` **nota** Si desea utilizar un sitio espejo que no sea el utilizado en este comando, puede elegir uno diferente en el sitio web de [Apache](https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.1/kafka_2.11-1.1.1.tgz). 1. Ejecute el siguiente comando en el directorio donde descargó el archivo TAR del paso anterior. ``` tar -xzf kafka_2.12-2.2.1.tgz ``` 1. Vaya al directorio **kafka\$12.12-2.2.1**. 1. ¿Abrir la consola Amazon MSK en [https://console.aws.amazon.com/msk/casa? region=us-east-1\$1/home/](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/). 1. En el panel izquierdo, seleccione **Clústeres** y, a continuación, elija el nombre `mkc-tutorial-cluster`. 1. Seleccione **Ver información del cliente**. 1. Copie la cadena de conexión de **texto sin formato**. 1. Seleccione **Listo**. 1. Ejecute el siguiente comando en la instancia del cliente (`mkc-tutorial-client`) y *bootstrapServerString* reemplácelo por el valor que guardó al ver la información del cliente del clúster. ``` /bin/kafka-topics.sh --create --bootstrap-server bootstrapServerString --replication-factor 2 --partitions 1 --topic mkc-tutorial-topic ``` Si el comando se ejecuta correctamente, verá el siguiente mensaje: `Created topic mkc-tutorial-topic.` **Paso siguiente** [Crea conector](mkc-create-connector.md) # Crea conector En este procedimiento, se describe cómo crear un conector desde la Consola de administración de AWS. **Creación del conector** 1. ¿Iniciar sesión en la Consola de administración de AWS consola Amazon MSK y abrirla desde [https://console.aws.amazon.com/msk/casa? region=us-east-1\$1/home/](https://console.aws.amazon.com/msk/home?region=us-east-1#/home/). 1. En el panel izquierdo, expanda **MSK Connect** y seleccione **Conectores**. 1. Elija **Crear conector**. 1. En la lista de complementos, seleccione `mkc-tutorial-plugin` y, a continuación, seleccione **Siguiente**. 1. Para el nombre del conector, ingrese `mkc-tutorial-connector`. 1. En la lista de clústeres, elija `mkc-tutorial-cluster`. 1. En la sección de **configuración de red del conector**, elija una de las siguientes opciones para el tipo de red: + **IPv4**(predeterminado): IPv4 solo para conectividad con destinos + **Doble pila**: para la conectividad a los destinos a través de ambos IPv4 y IPv6 (solo disponible si las subredes tienen IPv4 bloques IPv6 CIDR asociados a ellas) 1. Copie la siguiente configuración y péguela en el campo de configuración del conector. Asegúrese de reemplazar la región por el código del Región de AWS lugar donde va a crear el conector. Además, sustituya el nombre ** del bucket de Amazon S3 por el nombre de su bucket en el siguiente ejemplo. ``` connector.class=io.confluent.connect.s3.S3SinkConnector s3.region=us-east-1 format.class=io.confluent.connect.s3.format.json.JsonFormat flush.size=1 schema.compatibility=NONE tasks.max=2 topics=mkc-tutorial-topic partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner storage.class=io.confluent.connect.s3.storage.S3Storage s3.bucket.name= topics.dir=tutorial ``` 1. En **Permisos de acceso**, elija `mkc-tutorial-role`. 1. Elija **Siguiente**. En la página **Seguridad**, vuelva a seleccionar **Siguiente**. 1. En la página **Registros**, seleccione **Siguiente**. 1. En la página **Revisar y crear**, revise la configuración del conector y seleccione **Crear conector**. **Paso siguiente** [Envío de datos al clúster de MSK](mkc-send-data.md) # Envío de datos al clúster de MSK En este paso, envía los datos al tema de Apache Kafka que creó anteriormente y, a continuación, busca los mismos datos en el bucket de S3 de destino. **Envío de datos al clúster de MSK** 1. En la carpeta `bin` de la instalación de Apache Kafka en la instancia del cliente, cree un archivo de texto denominado `client.properties` con el siguiente contenido. ``` security.protocol=SASL_SSL sasl.mechanism=AWS_MSK_IAM ``` 1. Ejecute el siguiente comando para crear un productor de la consola. *BootstrapBrokerString*Sustitúyalo por el valor que obtuvo al ejecutar el comando anterior. ``` /bin/kafka-console-producer.sh --broker-list BootstrapBrokerString --producer.config client.properties --topic mkc-tutorial-topic ``` 1. Escriba el mensaje que desee y pulse **Enter (Entrar)**. Repita este paso dos o tres veces. Cada vez que introduzca una línea y pulse **Enter (Entrar)**, dicha línea se envía al clúster de Apache Kafka como un mensaje separado. 1. Busque en el bucket de Amazon S3 de los mensajes que envió en el paso anterior. # Descripción de los conectores Un conector integra sistemas externos y servicios de Amazon con Apache Kafka copiando continuamente los datos de streaming de un origen de datos a su clúster de Apache Kafka o copiando continuamente los datos de su clúster a un receptor de datos. Un conector también puede realizar una lógica ligera, como la transformación, la conversión de formato o el filtrado de datos antes de entregarlos a un destino. Los conectores de origen extraen datos de un origen de datos y los envían al clúster, mientras que los conectores de destino extraen datos del clúster y los envían a un receptor de datos. En el siguiente diagrama, se ilustra la arquitectura de un conector. Un proceso de trabajo es un proceso de máquina virtual Java (JVM) que ejecuta la lógica del conector. Cada proceso de trabajo crea un conjunto de tareas que se ejecutan en subprocesos paralelos y realizan el trabajo de copiar los datos. Las tareas no almacenan el estado y, por lo tanto, se pueden iniciar, detener o reiniciar en cualquier momento para proporcionar una canalización de datos flexible y escalable. ![\[Diagrama que muestra la arquitectura de un clúster de conectores.\]](http://docs.aws.amazon.com/es_es/msk/latest/developerguide/images/mkc-worker-architecture.png) # Descripción de la capacidad de los conectores La capacidad total de un conector depende del número de trabajadores que tenga el conector, así como del número de unidades MSK Connect (MCUs) por trabajador. Cada MCU representa 1 vCPU de computación y 4 GiB de memoria. La memoria de la MCU corresponde a la memoria total de una instancia de proceso de trabajo y no a la memoria en montón que está en uso. Los procesos de trabajo de MSK Connect consumen direcciones IP en las subredes proporcionadas por el cliente. Cada proceso de trabajo utiliza una dirección IP de una de las subredes proporcionadas por el cliente. Debe asegurarse de tener suficientes direcciones IP disponibles en las subredes proporcionadas en respuesta a una CreateConnector solicitud para tener en cuenta la capacidad especificada, especialmente al escalar automáticamente los conectores, donde la cantidad de trabajadores puede fluctuar. Para crear un conector, debe elegir uno de los dos modos de capacidad siguientes. + *Aprovisionado*: elija este modo si conoce los requisitos de capacidad del conector. Debe especificar dos valores: + El número de procesos de trabajo. + El número de por trabajador MCUs . + *Escalado automático*: elija este modo si los requisitos de capacidad del conector son variables o si no los conoce de antemano. Cuando utiliza el modo de escalado automático, Amazon MSK Connect anula la propiedad `tasks.max` del conector con un valor que es proporcional al número de trabajadores que se ejecutan en el conector y al número MCUs de trabajadores por trabajador. Debe especificar tres conjuntos de valores: + El número mínimo y máximo de procesos de trabajo. + Los porcentajes de reducción vertical y escalado horizontal para el uso de la CPU, que se determinan mediante la métrica `CpuUtilization`. Cuando la métrica `CpuUtilization` del conector supera el porcentaje de escalado horizontal, MSK Connect aumenta la cantidad de procesos de trabajo que trabajan en el conector. Cuando la métrica `CpuUtilization` cae por debajo del porcentaje de reducción horizontal, MSK Connect reduce la cantidad de procesos de trabajo. El número de procesos de trabajo siempre se mantiene dentro de los números mínimo y máximo que se especificaron al crear el conector. + El número de por trabajador. MCUs + (Opcional) *Recuento máximo de tareas de escalado automático*: el número máximo de tareas asignadas al conector durante las operaciones de escalado automático. Este parámetro le permite establecer un límite máximo en la creación de tareas, lo que proporciona un mayor control sobre la utilización de los recursos y el paralelismo en relación con las particiones temáticas de Kafka. Para obtener más información sobre los trabajadores[Descripción de los procesos de trabajo de MSK Connect](msk-connect-workers.md), consulte y para obtener más información sobre el número máximo de tareas de escalado automático, consulte. [Conozca el número máximo de tareas de escalado automático](msk-connect-max-autoscaling-task-count.md) Para obtener más información sobre las métricas de MSK Connect, consulte [Supervisión de Amazon MSK Connect](mkc-monitoring-overview.md). # Conozca el número máximo de tareas de escalado automático El `maxAutoscalingTaskCount` parámetro es un campo de capacidad opcional disponible para los conectores de escalado automático en Amazon MSK Connect. Este parámetro le permite establecer un límite máximo en el número máximo de tareas que se pueden crear durante las operaciones de escalado automático de los conectores, lo que proporciona un mayor control sobre la utilización y el rendimiento de los recursos. Cuando utiliza el modo de capacidad con escalado automático, Amazon MSK Connect anula automáticamente la propiedad `tasks.max` del conector con un valor proporcional al número de trabajadores MCUs y por trabajador. El `maxAutoscalingTaskCount` parámetro proporciona una opción configurable adicional para limitar el número máximo de tareas creadas para el conector. Esta capacidad resulta especialmente útil cuando se quiere controlar el nivel de paralelismo en relación con el número de particiones temáticas del clúster de Kafka. Al establecer este límite, puede optimizar el rendimiento y evitar una distribución ineficiente de las tareas que podría producirse cuando el recuento de tareas calculado automáticamente supere los requisitos de carga de trabajo. ## Requisitos de configuración El `maxAutoscalingTaskCount` parámetro debe cumplir los siguientes requisitos: ``` maxAutoscalingTaskCount ≥ maxWorkerCount ``` Este requisito garantiza una utilización eficiente de los recursos al mantener al menos una tarea por trabajador. El sistema aplica este mínimo para optimizar la funcionalidad del conector. Si lo especifica`maxAutoscalingTaskCount`, el límite se aplica inmediatamente después de la creación del conector y durante todos los eventos de escalado posteriores. A medida que el número de trabajadores aumenta o disminuye durante las operaciones de escalado automático, el sistema sigue respetando este límite. El `tasks.max` valor se ajusta proporcionalmente al número de trabajadores y MCUs por trabajador, pero nunca supera el valor configurado. `maxAutoscalingTaskCount` Si no especifica este parámetro, el conector utiliza el cálculo estándar sin ningún límite: `tasks.max = workerCount × mcuCount × tasksPerMcu` (donde tasksPerMcu es 2). ## Cuándo usar maxAutoscalingTask Count Considere la posibilidad de usarlo `maxAutoscalingTaskCount` en los siguientes escenarios: + *Recuento de particiones limitado*: si sus temas de Kafka tienen un número fijo de particiones inferior al recuento de tareas calculado automáticamente, establecer un límite impide la creación de tareas inactivas sin trabajo que realizar. + *Optimización del rendimiento*: cuando haya identificado que un recuento de tareas específico proporciona un rendimiento óptimo para su carga de trabajo, puede limitar el número máximo de tareas para mantener un rendimiento constante. + *Administración de recursos*: cuando desee controlar el paralelismo máximo y el consumo de recursos de su conector, independientemente del número de trabajadores que estén trabajando. ## Ejemplo Para un conector con la siguiente configuración: ``` minWorkerCount: 1 maxWorkerCount: 4 mcuCount: 8 maxAutoscalingTaskCount: 15 ``` Sin `maxAutoscalingTaskCount` él, si se escalara a 4 trabajadores, el conector crearía 64 tareas (4 trabajadores × 8 MCUs × 2 tareas por MCU). Si `maxAutoscalingTaskCount` se establece en 15, el conector crea solo 15 tareas, lo que puede ser más adecuado si el tema de Kafka tiene 15 particiones o menos. # Configure el tipo de red de doble pila Amazon MSK Connect admite el tipo de red de doble pila para los nuevos conectores. Con las redes de doble pila, sus conectores se pueden conectar a destinos a través de una y. IPv4 IPv6 Tenga en cuenta que la IPv6 conectividad solo está disponible en el modo de doble pila (IPv4 \$1 IPv6); no se IPv6 admiten redes exclusivas. De forma predeterminada, los conectores nuevos utilizan el tipo de IPv4 red. Para crear un conector con un tipo de red de doble pila, asegúrese de cumplir los requisitos previos que se describen en la siguiente sección. Tenga en cuenta que, una vez que haya creado un conector con un tipo de red de doble pila, no podrá modificar su tipo de red. Para cambiar los tipos de red, debe eliminar y volver a crear el conector. Amazon MSK Connect también admite la conectividad de puntos de conexión de la API de servicio a través de IPv6 y IPv4. Para utilizar la IPv6 conectividad para las llamadas a la API, debe utilizar los puntos de enlace de doble pila. Para obtener más información sobre los puntos de enlace del servicio MSK Connect, consulte los puntos de enlace y las cuotas de [Amazon MSK Connect.](https://docs.aws.amazon.com/general/latest/gr/msk-connect.html) ## Requisitos previos para utilizar un tipo de red de doble pila Antes de configurar el tipo de red de doble pila para los conectores, asegúrese de que todas las subredes que proporcione durante la creación del conector tengan ambos bloques IPv6 y IPv4 bloques CIDR asignados. ## Consideraciones sobre el uso de un tipo de red de doble pila + IPv6 Actualmente, el soporte solo está disponible en el modo de doble pila (IPv4 \$1 IPv6), no solo como IPv6 + Los conectores con doble pila se pueden conectar tanto IPv4 a los sistemas de datos MSK como IPv6 a los sistemas de datos Sink o Source + El tipo de red no se puede modificar después de crear el conector; debe eliminarlo y volver a crearlo para cambiar los tipos de red + Todas las subredes especificadas durante la creación del conector deben admitir la doble pila para que la creación del conector se realice correctamente con un tipo de red de doble pila + Si se utilizan subredes de doble pila pero no se especifica ningún tipo de red, el conector pasará a ser -only de forma predeterminada por motivos de compatibilidad con versiones anteriores IPv4 + En el caso de los conectores existentes, no puede actualizar el tipo de red; debe eliminar y volver a crear el conector para cambiar los tipos de red + El uso de redes de doble pila no implica costes adicionales # Creación de un conector de En este procedimiento, se describe cómo crear un conector desde la Consola de administración de AWS. **Crear un conector mediante el Consola de administración de AWS** 1. Abra la consola de Amazon MSK en [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/). 1. En el panel izquierdo, en **MSK Connect**, seleccione **Conectores**. 1. Elija **Crear conector**. 1. Puede elegir entre usar un complemento personalizado existente para crear el conector o crear primero un complemento personalizado nuevo. Para obtener información acerca de los complementos personalizados y cómo crearlos, consulte [Creación de complementos personalizados](msk-connect-plugins.md). En este procedimiento, supongamos que tiene un complemento personalizado que desea usar. En la lista de complementos personalizados, busque el que desee usar, seleccione la casilla situada a la izquierda y, a continuación, elija **Siguiente**. 1. Escriba un nombre y, opcionalmente, una descripción. 1. Haga clic en el clúster al que desea conectarse. 1. En la sección de **configuración de red del conector**, elija una de las siguientes opciones para el tipo de red: + **IPv4**(predeterminado): IPv4 solo para conectividad con destinos + **Doble pila**: para la conectividad a los destinos a través de ambos IPv4 y IPv6 (solo está disponible si las subredes tienen IPv4 bloques IPv6 CIDR asociados a ellas) 1. Especifique la configuración del conector. Los parámetros de configuración que debe especificar dependen del tipo de conector que desee crear. Sin embargo, algunos parámetros son comunes a todos los conectores, por ejemplo, los parámetros `connector.class` y `tasks.max`. A continuación, se muestra un ejemplo de configuración para el [conector de recepción de Amazon S3 de Confluent](https://www.confluent.io/hub/confluentinc/kafka-connect-s3). ``` connector.class=io.confluent.connect.s3.S3SinkConnector tasks.max=2 topics=my-example-topic s3.region=us-east-1 s3.bucket.name=amzn-s3-demo-bucket flush.size=1 storage.class=io.confluent.connect.s3.storage.S3Storage format.class=io.confluent.connect.s3.format.json.JsonFormat partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter schema.compatibility=NONE ``` 1. A continuación, debe configurar la capacidad del conector. Puede elegir entre dos modos de capacidad: aprovisionada y con escalado automático. Para obtener información sobre estas dos opciones, consulte [Descripción de la capacidad de los conectores](msk-connect-capacity.md). 1. (Opcional) En la sección **Recuento máximo de tareas de escalado automático, utilice el campo Recuento** máximo de tareas de escalado automático para introducir el número máximo de tareas que desea asignar al conector durante las operaciones de escalado automático. El valor debe ser al menos igual al número máximo de trabajadores. Si no especifica ningún valor, el conector utiliza el cálculo estándar sin ningún límite. Para obtener más información, consulte [Conozca el número máximo de tareas de escalado automático](msk-connect-max-autoscaling-task-count.md). 1. Elija la configuración de proceso de trabajo predeterminada o una configuración de proceso de trabajo personalizada. Para obtener información sobre la creación de configuraciones de procesos de trabajo personalizadas, consulte [Descripción de los procesos de trabajo de MSK Connect](msk-connect-workers.md). 1. A continuación, especifique el rol de ejecución del servicio. Debe ser una función de IAM que MSK Connect pueda asumir y que conceda al conector todos los permisos que necesita para acceder a los recursos necesarios AWS . Estos permisos dependen de la lógica del conector. Para obtener información acerca de cómo crear este rol, consulte [Descripción del rol de ejecución de servicios](msk-connect-service-execution-role.md). 1. Seleccione **Siguiente**, revise la información de seguridad y, a continuación, vuelva a seleccionar **Siguiente**. 1. Especifique las opciones de registro que desee y elija **Enviar**. Para obtener más información acerca del registro, consulte [Registro de MSK Connect](msk-connect-logging.md). 1. En la página **Revisar y crear**, revise la configuración del conector y seleccione **Crear** conector. Para usar la API MSK Connect para crear un conector, consulte [CreateConnector](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateConnector.html). Puede usar la API `UpdateConnector` para modificar la configuración del conector. Para obtener más información, consulte [Actualización de un conector](mkc-update-connector.md). # Actualización de un conector Este procedimiento describe cómo actualizar la configuración de un conector de MSK Connect existente mediante la Consola de administración de AWS. **Actualización de la configuración del conector mediante el Consola de administración de AWS** 1. Abra la consola de Amazon MSK en [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/). 1. En el panel izquierdo, en **MSK Connect**, seleccione **Conectores**. 1. Seleccione un conector existente. 1. Elija **Editar configuración del conector** 1. Actualice la configuración del conector. No puede anular el `connector.class` uso UpdateConnector de. El siguiente ejemplo muestra una configuración de ejemplo para el conector Confluent Amazon S3 Sink. ``` connector.class=io.confluent.connect.s3.S3SinkConnector tasks.max=2 topics=my-example-topic s3.region=us-east-1 s3.bucket.name=amzn-s3-demo-bucket flush.size=1 storage.class=io.confluent.connect.s3.storage.S3Storage format.class=io.confluent.connect.s3.format.json.JsonFormat partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter schema.compatibility=NONE ``` 1. Seleccione **Enviar**. 1. Luego puede supervisar el estado actual de la operación en la pestaña **Operaciones** del conector. Para usar la API MSK Connect para actualizar la configuración de un conector, consulte [UpdateConnector](https://docs.aws.amazon.com/MSKC/latest/mskc/API_UpdateConnector.html). # Conexión desde conectores Las siguientes prácticas recomendadas pueden mejorar el rendimiento de la conectividad a Amazon MSK Connect. ## No se superpongan IPs para el peering de Amazon VPC o Transit Gateway Si utiliza la interconexión de Amazon VPC o Transit Gateway con Amazon MSK Connect, no configure el conector para llegar a los recursos de VPC interconectados que se encuentren dentro de los rangos de CIDR: IPs + "10.99.0.0/16" + "192.168.0.0/16" + "172.21.0.0/16" # Creación de complementos personalizados Un complemento es un AWS recurso que contiene el código que define la lógica del conector. Debe cargar un archivo JAR (o un archivo ZIP que contenga uno o más archivos JAR) en un bucket de S3 y especificar la ubicación del bucket al crear el complemento. Cuando se crea el complemento, MSK Connect copia el contenido del objeto S3 en ese momento. No mantiene un enlace al objeto S3, por lo que cualquier modificación posterior del objeto no afectará al complemento ni a sus conectores. Al crear un conector, debe especificar el complemento que desea que MSK Connect utilice para este. La relación entre los complementos y los conectores es la one-to-many siguiente: puede crear uno o más conectores a partir del mismo complemento. **nota** Los complementos personalizados no se pueden actualizar in situ. Para usar una nueva versión del código del complemento, elimine todos los conectores que hagan referencia al complemento, elimine el complemento y, a continuación, vuelva a crearlo. **Paquete de dependencias para complementos personalizados** Te recomendamos que incluyas todos los archivos JAR y las dependencias necesarios para tu complemento. Package el conector como uno de los siguientes: Un archivo ZIP que contiene todos los archivos JAR y las dependencias necesarios para el complemento. Un único uber JAR que contiene todos los archivos de clases del complemento y sus dependencias. No agrupar las dependencias del plugin puede afectar a la disponibilidad o la compatibilidad en el entorno de ejecución y provocar errores inesperados. Para obtener información sobre cómo desarrollar el código de un conector, consulte la [Guía de desarrollo de conectores](https://kafka.apache.org/documentation/#connect_development) en la documentación de Apache Kafka. **Crear un complemento personalizado mediante el Consola de administración de AWS** 1. Abra la consola de Amazon MSK en [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/). 1. En el panel izquierdo, en **MSK Connect**, seleccione **Complementos personalizados**. 1. Seleccione **Crear complemento personalizado**. 1. Elija **Examinar S3**. 1. En la lista de buckets de S3, elija el bucket que contenga el archivo JAR o ZIP del complemento. 1. En la lista de objetos, seleccione la casilla situada a la izquierda del archivo JAR o ZIP del complemento y, a continuación, seleccione **Elegir**. 1. Seleccione **Crear complemento personalizado**. Para usar la API MSK Connect para crear un complemento personalizado, consulte [CreateCustomPlugin](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateCustomPlugin.html). # Descripción de los procesos de trabajo de MSK Connect Un proceso de trabajo es un proceso de máquina virtual Java (JVM) que ejecuta la lógica del conector. Cada proceso de trabajo crea un conjunto de tareas que se ejecutan en subprocesos paralelos y realizan el trabajo de copiar los datos. Las tareas no almacenan el estado y, por lo tanto, se pueden iniciar, detener o reiniciar en cualquier momento para proporcionar una canalización de datos flexible y escalable. Los demás procesos de trabajo detectan automáticamente los cambios en el número de procesos de trabajo, ya sea debido a un problema de escalamiento o a fallos inesperados. Se coordinan para reequilibrar las tareas entre el conjunto de procesos de trabajo restantes. Los procesos de trabajo de Connect utilizan los grupos de consumidores de Apache Kafka para coordinarse y reequilibrarse. Si los requisitos de capacidad de su conector son variables o difíciles de estimar, puede dejar que MSK Connect escale el número de procesos de trabajo según sea necesario entre el límite inferior y el límite superior que especifique. Como alternativa, puede especificar el número exacto de procesos de trabajo en los que desea ejecutar la lógica del conector. Para obtener más información, consulte [Descripción de la capacidad de los conectores](msk-connect-capacity.md). **Los procesos de trabajo de MSK Connect consumen direcciones IP** Los procesos de trabajo de MSK Connect consumen direcciones IP en las subredes proporcionadas por el cliente. Cada proceso de trabajo utiliza una dirección IP de una de las subredes proporcionadas por el cliente. Debe asegurarse de tener suficientes direcciones IP disponibles en las subredes proporcionadas a una CreateConnector solicitud para tener en cuenta la capacidad especificada, especialmente al escalar automáticamente los conectores, donde la cantidad de trabajadores puede fluctuar. ## Configuración predeterminada del proceso de trabajo MSK Connect proporciona la siguiente configuración de proceso de trabajo predeterminada: ``` key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter ``` # Propiedades de configuración de proceso de trabajo compatibles MSK Connect proporciona una configuración de proceso de trabajo predeterminada. También tiene la opción de crear una configuración de proceso de trabajo personalizada para utilizarla con sus conectores. La siguiente lista incluye información sobre las propiedades de configuración de proceso de trabajo que Amazon MSK Connect admite o no admite. + Solo se necesitan las propiedades `key.converter` y `value.converter`. + MSK Connect admite las siguientes propiedades de configuración de `producer.`. ``` producer.acks producer.batch.size producer.buffer.memory producer.compression.type producer.enable.idempotence producer.key.serializer producer.linger.ms producer.max.request.size producer.metadata.max.age.ms producer.metadata.max.idle.ms producer.partitioner.class producer.reconnect.backoff.max.ms producer.reconnect.backoff.ms producer.request.timeout.ms producer.retry.backoff.ms producer.value.serializer ``` + MSK Connect admite las siguientes propiedades de configuración de `consumer.`. ``` consumer.allow.auto.create.topics consumer.auto.offset.reset consumer.check.crcs consumer.fetch.max.bytes consumer.fetch.max.wait.ms consumer.fetch.min.bytes consumer.heartbeat.interval.ms consumer.key.deserializer consumer.max.partition.fetch.bytes consumer.max.poll.interval.ms consumer.max.poll.records consumer.metadata.max.age.ms consumer.partition.assignment.strategy consumer.reconnect.backoff.max.ms consumer.reconnect.backoff.ms consumer.request.timeout.ms consumer.retry.backoff.ms consumer.session.timeout.ms consumer.value.deserializer ``` + Se admiten todas las demás propiedades de configuración que no comiencen por los prefijos `producer.` o `consumer.`, *excepto* las siguientes propiedades. ``` access.control. admin. admin.listeners.https. client. connect. inter.worker. internal. listeners.https. metrics. metrics.context. rest. sasl. security. socket. ssl. topic.tracking. worker. bootstrap.servers config.storage.topic connections.max.idle.ms connector.client.config.override.policy group.id listeners metric.reporters plugin.path receive.buffer.bytes response.http.headers.config scheduled.rebalance.max.delay.ms send.buffer.bytes status.storage.topic ``` Para obtener más información sobre las propiedades de configuración de trabajo y lo que representan, consulte [Kafka Connect Configs](https://kafka.apache.org/documentation/#connectconfigs) en la documentación de Apache Kafka. # Creación de una configuración de proceso de trabajo personalizada En este procedimiento, se describe cómo crear una configuración de proceso de trabajo personalizada utilizando la Consola de administración de AWS. **Crear una configuración de trabajo personalizada mediante el Consola de administración de AWS** 1. Abra la consola de Amazon MSK en [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/). 1. En el panel izquierdo, en **MSK Connect**, seleccione **Configuraciones de proceso de trabajo**. 1. Elija **Crear configuración de proceso de trabajo**. 1. Introduzca un nombre y una descripción opcional y, a continuación, añada las propiedades y los valores en los que desee establecerlos. 1. Elija **Crear configuración de proceso de trabajo**. Para usar la API MSK Connect para crear una configuración de trabajo, consulte [CreateWorkerConfiguration](https://docs.aws.amazon.com/MSKC/latest/mskc/API_CreateWorkerConfiguration.html). # Administración de los desplazamientos de los conectores de origen con `offset.storage.topic` En esta sección se proporciona información que le ayudará a gestionar los desplazamientos de los conectores de origen mediante el tema de *almacenamiento de desplazamientos*. El tema del almacenamiento de desplazamientos es un tema interno que Kafka Connect utiliza para almacenar los desplazamientos de configuración de conectores y tareas. ## Consideraciones Tenga en cuenta lo siguiente cuando gestione los desplazamientos del conector de origen. + Para especificar un tema de almacenamiento de desplazamientos, proporcione el nombre del tema de Kafka en el que se almacenan los desplazamientos de los conectores como valor `offset.storage.topic` en su configuración de proceso de trabajo. + Tenga cuidado al realizar cambios en la configuración de un conector. El cambio de los valores de configuración puede provocar un comportamiento no deseado del conector si un conector de origen utiliza valores de la configuración para introducir registros de desplazamiento. Recomendamos que consulte la documentación de su complemento para obtener orientación. + **Personalice el número predeterminado de particiones**: además de personalizar la configuración de proceso de trabajo añadiendo `offset.storage.topic`, puede personalizar el número de particiones para los temas de compensación y almacenamiento de estado. Las particiones predeterminadas para los temas internos son las siguientes. + `config.storage.topic`: 1, no configurable, debe ser un tema de partición única + `offset.storage.topic`: 25, configurable proporcionando `offset.storage.partitions` + `status.storage.topic`: 5, configurable proporcionando `status.storage.partitions` + **Eliminación manual de temas**: Amazon MSK Connect crea nuevos temas internos de Kafka Connect (el nombre del tema comienza por `__amazon_msk_connect`) en cada implementación de conectores. Los temas antiguos que se adjuntan a los conectores eliminados no se eliminan automáticamente, ya que los temas internos, como `offset.storage.topic`, se pueden reutilizar entre los conectores. Sin embargo, puede eliminar manualmente los temas internos no utilizados creados por MSK Connect. Los temas internos se nombran siguiendo el formato `__amazon_msk_connect__connector_name_connector_id`. La expresión regular `__amazon_msk_connect__connector_name_connector_id` se puede utilizar para eliminar los temas internos. No debe eliminar un tema interno que utiliza actualmente un conector en ejecución. + **Uso del mismo nombre para los temas internos creados por MSK Connect**: si desea reutilizar el tema de almacenamiento de desplazamientos para consumir los desplazamientos de un conector creado anteriormente, debe asignar al nuevo conector el mismo nombre que al conector anterior. La propiedad `offset.storage.topic` se puede establecer mediante la configuración del proceso de trabajo para asignar el mismo nombre al conector `offset.storage.topic` y reutilizarla entre distintos conectores. Esta configuración se describe en [Gestión de desplazamientos de los conectores](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-create-custom-worker-config). MSK Connect no permite que diferentes conectores compartan `config.storage.topic` y `status.storage.topic`. Estos temas se crean cada vez que se crea un conector nuevo en MSKC. Se les asigna un nombre automáticamente según el formato `__amazon_msk_connect__connector_name_connector_id` y, por lo tanto, son diferentes en los distintos conectores que cree. # Uso del tema de almacenamiento de desplazamiento predeterminado De forma predeterminada, Amazon MSK Connect genera un nuevo tema de almacenamiento de desplazamiento en el clúster de Kafka para cada conector que cree. MSK crea el nombre del tema por defecto utilizando partes del ARN del conector. Por ejemplo, `__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2`. # Uso del tema de almacenamiento de desplazamiento personalizado Para proporcionar continuidad de desplazamiento entre los conectores de origen, puede utilizar un tema de almacenamiento de desplazamiento de su elección en lugar del tema predeterminado. Especificar un tema de almacenamiento de desplazamiento le ayuda a realizar tareas como crear un conector de origen que reanude la lectura desde el último desplazamiento de un conector anterior. Para especificar un tema de almacenamiento de desplazamiento, debe proporcionar un valor para la propiedad `offset.storage.topic` en su configuración de proceso de trabajo antes de crear un conector. Si desea reutilizar el tema de almacenamiento de desplazamientos para consumir los desplazamientos de un conector creado anteriormente, debe asignar al nuevo conector el mismo nombre que al conector anterior. Si crea un tema de almacenamiento de desplazamiento personalizado, debe definir [https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy](https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy) como `compact` en la configuración del tema. **nota** Si especifica un tema de almacenamiento de desplazamiento al crear un conector de *recepción*, MSK Connect crea el tema si aún no existe. Sin embargo, el tema no se utilizará para almacenar los desplazamientos de los conectores. En cambio, los desplazamientos de los conectores de recepción se gestionan mediante el protocolo de grupos de consumidores de Kafka. Cada conector de recepción crea un grupo denominado `connect-{CONNECTOR_NAME}`. Mientras exista el grupo de consumidores, cualquier conector de recepción sucesivo que se cree con el mismo valor `CONNECTOR_NAME` se mantendrá desde el último desplazamiento asignado. **importante** Si desea actualizar la configuración de un conector existente y, al mismo tiempo, mantener la continuidad de la compensación, utilice la UpdateConnector API. Para obtener más información, consulte [Actualización de un conector](mkc-update-connector.md). **Example : Especificar un tema de almacenamiento de compensación al recrear un conector de origen** Si necesita eliminar y volver a crear un conector manteniendo la continuidad de la compensación, puede especificar un tema de almacenamiento de la compensación en su configuración de trabajo. Por ejemplo, supongamos que tiene un conector de captura de datos de cambios (CDC) y quiere volver a crearlo sin perder su lugar en la transmisión de CDC. En los siguientes pasos se muestra cómo realizar esta tarea. 1. En el equipo cliente, ejecute el siguiente comando para buscar el nombre del tema de almacenamiento de desplazamiento del conector. Sustituya `` por la cadena de agente de arranque de su clúster. Para ver instrucciones sobre cómo obtener la cadena de su agente de arranque, consulte [Obtención de agentes de arranque para un clúster de Amazon MSK](msk-get-bootstrap-brokers.md). ``` /bin/kafka-topics.sh --list --bootstrap-server ``` El siguiente resultado muestra una lista de todos los temas del clúster, incluidos los temas de conectores internos predeterminados. En este ejemplo, el conector CDC existente utiliza el [tema de almacenamiento de desplazamiento predeterminado](msk-connect-default-offset-storage-topic.md) creado por MSK Connect. Por eso el tema del almacenamiento de desplazamiento se denomina `__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2`. ``` __consumer_offsets __amazon_msk_canary __amazon_msk_connect_configs_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2 __amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2 __amazon_msk_connect_status_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2 my-msk-topic-1 my-msk-topic-2 ``` 1. Abra la consola de Amazon MSK en [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk). 1. Elija el conector de la lista **Conectores**. Copie y guarde el contenido del campo de **Configuración del conector** para poder modificarlo y usarlo para crear el nuevo conector. 1. Para eliminar el conector, elija **Eliminar**. A continuación, ingrese el nombre del conector en el campo de entrada de texto para confirmar la eliminación. 1. Cree una configuración de proceso de trabajo personalizada con valores que se adapten a su caso de uso. Para obtener instrucciones, consulte [Creación de una configuración de proceso de trabajo personalizada](msk-connect-create-custom-worker-config.md). En su configuración de proceso de trabajo, debe especificar el nombre del tema de almacenamiento de desplazamiento que ha recuperado anteriormente como valor para `offset.storage.topic` como en la siguiente configuración. ``` config.providers.secretManager.param.aws.region=eu-west-3 key.converter= value.converter= config.providers.secretManager.class=com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider config.providers=secretManager offset.storage.topic=__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2 ``` 1. **importante** Debe asignar al conector nuevo el mismo nombre que al conector anterior. Cree un conector nuevo con la configuración de proceso de trabajo que configuró en el paso anterior. Para obtener instrucciones, consulte [Creación de un conector de](mkc-create-connector-intro.md). # Tutorial: Externalización de la información confidencial con proveedores de configuración En este ejemplo, se muestra cómo externalizar la información confidencial de Amazon MSK Connect mediante un proveedor de configuración de código abierto. Un proveedor de configuración le permite especificar variables en lugar de texto sin formato en una configuración de conector o de trabajo, y los procesos de trabajo que se ejecutan en su conector resuelven estas variables en tiempo de ejecución. Esto evita que las credenciales y otros secretos se almacenen en texto sin formato. El proveedor de configuración del ejemplo admite la recuperación de los parámetros de configuración de AWS Secrets Manager, Amazon S3 y Systems Manager (SSM). En el [paso 2](#msk-connect-config-providers), puede ver cómo configurar el almacenamiento y la recuperación de información confidencial para el servicio que desee configurar. ## Consideraciones Tenga en cuenta lo siguiente al utilizar el proveedor de configuración de MSK con Amazon MSK Connect: + Al utilizar los proveedores de configuración, asigne los permisos adecuados al rol de ejecución del servicio de IAM. + Defina los proveedores de configuración en las configuraciones de proceso de trabajo y su implementación en la configuración del conector. + Los valores de configuración confidenciales pueden aparecer en los registros de los conectores si un complemento no los define como secretos. Kafka Connect trata los valores de configuración indefinidos de la misma manera que cualquier otro valor de texto sin formato. Para obtener más información, consulte [Impedir que los secretos aparezcan en los registros de los conectores](msk-connect-logging.md#msk-connect-logging-secrets). + De forma predeterminada, MSK Connect reinicia con frecuencia un conector cuando este utiliza un proveedor de configuración. Para desactivar este comportamiento de reinicio, puede establecer el valor de `config.action.reload` en `none` en la configuración del conector. ## Creación de un complemento personalizado y carga a S3 Para crear un complemento personalizado, cree un archivo zip que contenga el conector y ejecute msk-config-provider los siguientes comandos en su máquina local. **Para crear un complemento personalizado utilizando una ventana de terminal y Debezium como conector** Utilice la AWS CLI para ejecutar comandos como superusuario con credenciales que le permitan acceder a su bucket de AWS S3. Para obtener información sobre la instalación y configuración de la AWS CLI, consulte [Introducción a la AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html) en la *Guía del AWS Command Line Interface usuario*. Para obtener información sobre el uso de la AWS CLI con Amazon S3, consulte [Uso de Amazon S3 con la AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-services-s3.html) en la *Guía del AWS Command Line Interface usuario*. 1. En una ventana de terminal, cree una carpeta denominada `custom-plugin` en su espacio de trabajo mediante el siguiente comando. ``` mkdir custom-plugin && cd custom-plugin ``` 1. Descargue la última versión estable del **complemento MySQL Connector** desde el [sitio de Debezium](https://debezium.io/releases/) mediante el siguiente comando. ``` wget https://repo1.maven.org/maven2/io/debezium/debezium-connectormysql/ 2.2.0.Final/debezium-connector-mysql-2.2.0.Final-plugin.tar.gz ``` Extraiga el archivo gzip descargado de la carpeta `custom-plugin` con el siguiente comando. ``` tar xzf debezium-connector-mysql-2.2.0.Final-plugin.tar.gz ``` 1. Descargue el [archivo zip del proveedor de configuración de MSK](https://github.com/aws-samples/msk-config-providers/releases/download/r0.4.0/msk-config-providers-0.4.0-with-dependencies.zip) con el siguiente comando. ``` wget https://github.com/aws-samples/msk-config-providers/releases/download/r0.4.0/msk-config-providers-0.4.0-with-dependencies.zip ``` Extraiga el archivo zip descargado de la carpeta `custom-plugin` con el siguiente comando. ``` unzip msk-config-providers-0.4.0-with-dependencies.zip ``` 1. Comprima el contenido del proveedor de configuración de MSK del paso anterior y el conector personalizado en un único archivo denominado `custom-plugin.zip`. ``` zip -r ../custom-plugin.zip * ``` 1. Cargue el archivo en S3 para consultarlo más adelante. ``` aws s3 cp ../custom-plugin.zip s3: ``` 1. En la consola de Amazon MSK, en la sección **MSK Connect**, elija **Custom Plugin**, elija **Create custom plugin** y explore el bucket **s3: < *S3\$1URI\$1BUCKET\$1LOCATION* >** S3 para seleccionar el archivo ZIP del plugin personalizado que acaba de cargar. ![\[Amazon S3 bucket interface showing a single custom-plugin.zip file in the debezium folder.\]](http://docs.aws.amazon.com/es_es/msk/latest/developerguide/images/s3-object-browser.png) 1. Introduzca **debezium-custom-plugin** para el nombre del complemento. De manera opcional, ingrese una descripción y elija **Crear complemento personalizado**. ![\[Amazon S3 bucket interface showing a single custom-plugin.zip file in the debezium folder.\]](http://docs.aws.amazon.com/es_es/msk/latest/developerguide/images/create-custom-plugin.png) ## Configuración de los parámetros y permisos para los distintos proveedores Puede configurar los valores de los parámetros en estos tres servicios: + Secrets Manager  + Almacén de parámetros de Systems Manager + S3 - Simple Storage Service Elija una de las pestañas siguientes para consultar las instrucciones sobre la configuración de parámetros y los permisos pertinentes para ese servicio. ------ #### [ Configure in Secrets Manager ] **Configuración de valores de parámetros en Secrets Manager** 1. Abra la [consola de Secrets Manager](https://console.aws.amazon.com/secretsmanager/). 1. Cree un nuevo secreto para almacenar sus credenciales o secretos. Para obtener instrucciones, consulte [Crear un AWS Secrets Manager secreto](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html) en la Guía del *AWS Secrets Manager usuario*. 1. Copie el ARN de su secreto. 1. Añada los permisos de Secrets Manager de la siguiente política de ejemplo al [rol de ejecución del servicio](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). Reemplace el ARN de ejemplo, `arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234`, por el ARN del secreto. 1. Agregue la configuración de proceso de trabajo y las instrucciones del conector. **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "secretsmanager:GetResourcePolicy", "secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret", "secretsmanager:ListSecretVersionIds" ], "Resource": [ "arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234" ] } ] } ``` 1. Para utilizar el proveedor de configuración Secrets Manager, copie las siguientes líneas de código en el cuadro de texto de configuración del proceso de trabajo del paso 3: ``` # define name of config provider: config.providers = secretsmanager # provide implementation classes for secrets manager: config.providers.secretsmanager.class = com.amazonaws.kafka.config.providers.SecretsManagerConfigProvider # configure a config provider (if it needs additional initialization), for example you can provide a region where the secrets or parameters are located: config.providers.secretsmanager.param.region = us-east-1 ``` 1. Para el proveedor de configuración Secrets Manager, copie las siguientes líneas de código en la configuración del conector del paso 4. ``` #Example implementation for secrets manager variable database.user=${secretsmanager:MSKAuroraDBCredentials:username} database.password=${secretsmanager:MSKAuroraDBCredentials:password} ``` También puede utilizar el paso anterior con más proveedores de configuración. ------ #### [ Configure in Systems Manager Parameter Store ] **Configuración de valores de parámetros en el almacén de parámetros de Systems Manager** 1. Abra la consola de [Systems Manager](https://console.aws.amazon.com/systems-manager/). 1. En el panel de navegación, elija **Parameter Store (Almacén de parámetros)**. 1. Cree un parámetro nuevo para guardarlo en Systems Manager. Para obtener instrucciones, consulte [Creación de un parámetro de Systems Manager (consola)](https://docs.aws.amazon.com/systems-manager/latest/userguide/parameter-create-console.html) en la Guía del AWS Systems Manager usuario. 1. Copie el ARN de su parámetro. 1. Añada los permisos de Systems Manager de la siguiente política de ejemplo al [rol de ejecución del servicio](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). **Sustitúyalo por el ARN de su parámetro. **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "VisualEditor0", "Effect": "Allow", "Action": [ "ssm:GetParameterHistory", "ssm:GetParametersByPath", "ssm:GetParameters", "ssm:GetParameter" ], "Resource": "arn:aws:ssm:us-east-1:123456789000:parameter/MyParameterName" } ] } ``` 1. Para utilizar el proveedor de configuración del almacén de parámetros, copie las siguientes líneas de código en el cuadro de texto de configuración del proceso de trabajo del paso 3: ``` # define name of config provider: config.providers = ssm # provide implementation classes for parameter store: config.providers.ssm.class = com.amazonaws.kafka.config.providers.SsmParamStoreConfigProvider # configure a config provider (if it needs additional initialization), for example you can provide a region where the secrets or parameters are located: config.providers.ssm.param.region = us-east-1 ``` 1. Para el proveedor de configuración del almacén de parámetros, copie las siguientes líneas de código en la configuración del conector del paso 5. ``` #Example implementation for parameter store variable schema.history.internal.kafka.bootstrap.servers=${ssm::MSKBootstrapServerAddress} ``` También puede agrupar los dos pasos anteriores con más proveedores de configuración. ------ #### [ Configure in Amazon S3 ] **Para configurar objects/files en Amazon S3** 1. Abra la [consola de Amazon S3](https://console.aws.amazon.com/s3/). 1. Cargue su objeto en un bucket en S3. Para obtener instrucciones, consulte [Carga de objetos](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html). 1. Copie el ARN de su objeto. 1. Añada los permisos de lectura de objetos de Amazon S3 de la siguiente política de ejemplo al [rol de ejecución del servicio](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). Reemplace el ARN de ejemplo, `arn:aws:s3:::`, por el ARN del objeto. **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "VisualEditor0", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::" } ] } ``` 1. Para utilizar el proveedor de configuración de Amazon S3, copie las siguientes líneas de código en el cuadro de texto de configuración del proceso de trabajo del paso 3: ``` # define name of config provider: config.providers = s3import # provide implementation classes for S3: config.providers.s3import.class = com.amazonaws.kafka.config.providers.S3ImportConfigProvider ``` 1. Para el proveedor de configuración de Amazon S3, copie las siguientes líneas de código en la configuración del conector del paso 4. ``` #Example implementation for S3 object database.ssl.truststore.location = ${s3import:us-west-2:my_cert_bucket/path/to/trustore_unique_filename.jks} ``` También puede agrupar los dos pasos anteriores con más proveedores de configuración. ------ ## Creación de una configuración de proceso de trabajo personalizada con información sobre su proveedor de configuración 1. Seleccione **Configuraciones de proceso de trabajo** en la sección **Amazon MSK Connect**. 1. Seleccione **Crear configuración de proceso de trabajo**. 1. Introduzca `SourceDebeziumCustomConfig` en el cuadro de texto Nombre de configuración de proceso de trabajo. La descripción es opcional. 1. Copie el código de configuración correspondiente en función de los proveedores que desee y péguelo en el cuadro de texto **Configuración del proceso de trabajo**. 1. Este es un ejemplo de la configuración de proceso de trabajo para los tres proveedores: ``` key.converter=org.apache.kafka.connect.storage.StringConverter key.converter.schemas.enable=false value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false offset.storage.topic=offsets_my_debezium_source_connector # define names of config providers: config.providers=secretsmanager,ssm,s3import # provide implementation classes for each provider: config.providers.secretsmanager.class = com.amazonaws.kafka.config.providers.SecretsManagerConfigProvider config.providers.ssm.class = com.amazonaws.kafka.config.providers.SsmParamStoreConfigProvider config.providers.s3import.class = com.amazonaws.kafka.config.providers.S3ImportConfigProvider # configure a config provider (if it needs additional initialization), for example you can provide a region where the secrets or parameters are located: config.providers.secretsmanager.param.region = us-east-1 config.providers.ssm.param.region = us-east-1 ``` 1. Haga clic en Crear configuración del proceso de trabajo. ## Creación del conector 1. Cree un conector nuevo siguiendo las instrucciones de [Crear un conector nuevo](https://docs.aws.amazon.com/msk/latest/developerguide/mkc-create-connector.html). 1. Elija el archivo `custom-plugin.zip` que cargó en su bucket de S3 en [Creación de un complemento personalizado y carga a S3](#msk-connect-config-providers-create-custom-plugin) como origen del complemento personalizado. 1. Copie el código de configuración correspondiente en función de los proveedores que desee y péguelo en el cuadro de texto Configuración del conector. 1. Este es un ejemplo de la configuración del conector para los tres proveedores: ``` #Example implementation for parameter store variable schema.history.internal.kafka.bootstrap.servers=${ssm::MSKBootstrapServerAddress} #Example implementation for secrets manager variable database.user=${secretsmanager:MSKAuroraDBCredentials:username} database.password=${secretsmanager:MSKAuroraDBCredentials:password} #Example implementation for Amazon S3 file/object database.ssl.truststore.location = ${s3import:us-west-2:my_cert_bucket/path/to/trustore_unique_filename.jks} ``` 1. Seleccione **Usar una configuración personalizada** y elija **SourceDebeziumCustomConfig**una opción en el menú desplegable **Worker Configuration**. 1. Siga los pasos restantes de las instrucciones de [Crear un conector](https://docs.aws.amazon.com/msk/latest/developerguide/mkc-create-connector.html). # Políticas y roles de IAM para MSK Connect Esta sección le ayuda a configurar las políticas y funciones de IAM adecuadas para implementar y gestionar Amazon MSK Connect de forma segura en su AWS entorno. En las secciones siguientes, se explica el rol de ejecución del servicio que se debe utilizar con MSK Connect, incluida la política de confianza requerida y los permisos adicionales necesarios al conectarse a un clúster de MSK autenticado con IAM. La página también proporciona ejemplos de políticas integrales de IAM para garantizar el acceso total a la funcionalidad de MSK Connect, así como detalles sobre las políticas AWS gestionadas disponibles para el servicio. **Topics** + [Descripción del rol de ejecución de servicios](msk-connect-service-execution-role.md) + [Ejemplo de política de IAM para MSK Connect](mkc-iam-policy-examples.md) + [Prevención de los problemas del suplente confuso entre servicios](cross-service-confused-deputy-prevention.md) + [AWS políticas administradas para MSK Connect](mkc-security-iam-awsmanpol.md) + [Uso de roles vinculados a servicios para MSK Connect](mkc-using-service-linked-roles.md) # Descripción del rol de ejecución de servicios **nota** Amazon MSK Connect no admite el uso del [rol vinculado al servicio](mkc-using-service-linked-roles.md) como rol de ejecución del servicio. Debe crear un rol de ejecución del servicio independiente. Para obtener instrucciones sobre cómo crear un rol de IAM personalizado, consulte [Crear un rol para delegar permisos a un AWS servicio](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-service.html) en la Guía del usuario de *IAM*. Al crear un conector con MSK Connect, debe especificar un rol AWS Identity and Access Management (IAM) para usarlo con él. Su rol de ejecución del servicio debe tener la siguiente política de confianza para que MSK Connect pueda asumirla. Para obtener información sobre las claves de contexto de condición en esta política, consulte [Prevención de los problemas del suplente confuso entre servicios](cross-service-confused-deputy-prevention.md). ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "kafkaconnect.amazonaws.com" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "aws:SourceAccount": "123456789012" }, "ArnLike": { "aws:SourceArn": "arn:aws:kafkaconnect:us-east-1:123456789012:connector/myConnector/abc12345-abcd-4444-a8b9-123456f513ed-2" } } } ] } ``` ------ Si el clúster de Amazon MSK que desea utilizar con su conector es un clúster que utiliza la autenticación de IAM, debe añadir la siguiente política de permisos al rol de ejecución del servicio del conector. Para obtener información sobre cómo encontrar el UUID del clúster y cómo crear un tema ARNs, consulte. [Recursos de la política de autorización](kafka-actions.md#msk-iam-resources) ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kafka-cluster:Connect", "kafka-cluster:DescribeCluster" ], "Resource": [ "arn:aws:kafka:us-east-1:000000000001:cluster/testClusterName/300d0000-0000-0005-000f-00000000000b-1" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:ReadData", "kafka-cluster:DescribeTopic" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:topic/myCluster/300a0000-0000-0003-000a-00000000000b-6/__amazon_msk_connect_read" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:WriteData", "kafka-cluster:DescribeTopic" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:topic/testCluster/300f0000-0000-0008-000d-00000000000m-7/__amazon_msk_connect_write" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:CreateTopic", "kafka-cluster:WriteData", "kafka-cluster:ReadData", "kafka-cluster:DescribeTopic" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:topic/testCluster/300f0000-0000-0008-000d-00000000000m-7/__amazon_msk_connect_*" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:AlterGroup", "kafka-cluster:DescribeGroup" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:group/testCluster/300d0000-0000-0005-000f-00000000000b-1/__amazon_msk_connect_*", "arn:aws:kafka:us-east-1:123456789012:group/testCluster/300d0000-0000-0005-000f-00000000000b-1/connect-*" ] } ] } ``` ------ Según el tipo de conector, es posible que también necesites adjuntar a la función de ejecución del servicio una política de permisos que le permita acceder a AWS los recursos. Por ejemplo, si el conector necesita enviar datos a un bucket de S3, el rol de ejecución del servicio debe tener una política de permisos que conceda permiso para escribir en ese bucket. Para realizar pruebas, puede usar una de las políticas de IAM prediseñadas que otorgan acceso total, como `arn:aws:iam::aws:policy/AmazonS3FullAccess`. Sin embargo, por motivos de seguridad, le recomendamos que utilice la política más restrictiva que permita al conector leer desde la AWS fuente o escribir en el AWS receptor. # Ejemplo de política de IAM para MSK Connect Para dar a un usuario que no sea administrador acceso completo a todas las funciones de MSK Connect, adjunte una política como la siguiente al rol de IAM del usuario. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "MSKConnectFullAccess", "Effect": "Allow", "Action": [ "kafkaconnect:CreateConnector", "kafkaconnect:DeleteConnector", "kafkaconnect:DescribeConnector", "kafkaconnect:ListConnectors", "kafkaconnect:UpdateConnector", "kafkaconnect:CreateCustomPlugin", "kafkaconnect:DeleteCustomPlugin", "kafkaconnect:DescribeCustomPlugin", "kafkaconnect:ListCustomPlugins", "kafkaconnect:CreateWorkerConfiguration", "kafkaconnect:DeleteWorkerConfiguration", "kafkaconnect:DescribeWorkerConfiguration", "kafkaconnect:ListWorkerConfigurations" ], "Resource": "*" }, { "Sid": "IAMPassRole", "Effect": "Allow", "Action": "iam:PassRole", "Resource": "arn:aws:iam::123456789012:role/MSKConnectServiceRole", "Condition": { "StringEquals": { "iam:PassedToService": "kafkaconnect.amazonaws.com" } } }, { "Sid": "EC2NetworkAccess", "Effect": "Allow", "Action": [ "ec2:CreateNetworkInterface", "ec2:DescribeNetworkInterfaces", "ec2:DeleteNetworkInterface", "ec2:DescribeVpcs", "ec2:DescribeSubnets", "ec2:DescribeSecurityGroups" ], "Resource": "*" }, { "Sid": "MSKClusterAccess", "Effect": "Allow", "Action": [ "kafka:DescribeCluster", "kafka:DescribeClusterV2", "kafka:GetBootstrapBrokers" ], "Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/myCluster/" }, { "Sid": "MSKLogGroupAccess", "Effect": "Allow", "Action": [ "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents", "logs:DescribeLogStreams", "logs:DescribeLogGroups" ], "Resource": [ "arn:aws:logs:us-east-1:123456789012:log-group:/aws/msk-connect/*" ] }, { "Sid": "S3PluginAccess", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket", "s3:PutObject" ], "Resource": [ "arn:aws:s3:::amzn-s3-demo-bucket1-custom-plugins", "arn:aws:s3:::amzn-s3-demo-bucket1-custom-plugins/*" ] } ] } ``` ------ # Prevención de los problemas del suplente confuso entre servicios El problema de la sustitución confusa es un problema de seguridad en el que una entidad que no tiene permiso para realizar una acción puede obligar a una entidad con más privilegios a realizar la acción. En AWS, la suplantación de identidad entre servicios puede provocar un confuso problema de diputado. La suplantación entre servicios puede producirse cuando un servicio (el *servicio que lleva a cabo las llamadas*) llama a otro servicio (el *servicio al que se llama*). El servicio que lleva a cabo las llamadas se puedes manipular para utilizar sus permisos a fin de actuar en función de los recursos de otro cliente de una manera en la que no debe tener permiso para acceder. Para evitarlo, AWS proporciona herramientas que lo ayudan a proteger sus datos para todos los servicios con entidades principales de servicio a las que se les ha dado acceso a los recursos de su cuenta. Se recomienda utilizar las claves de contexto de condición global [https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourcearn](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourcearn) y [https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourceaccount](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourceaccount) en las políticas de recursos para limitar los permisos que MSK Connect concede a otro servicio para el recurso. Si el valor de `aws:SourceArn` no contiene el ID de cuenta (por ejemplo, un ARN de bucket de Amazon S3 no contiene el ID de cuenta), debe utilizar ambas claves de contexto de condición global para limitar los permisos. Si utiliza claves de contexto de condición global y el valor de `aws:SourceArn` contiene el ID de cuenta, el valor de `aws:SourceAccount` y la cuenta en el valor de `aws:SourceArn` deben utilizar el mismo ID de cuenta cuando se utiliza en la misma instrucción de política. Utiliza `aws:SourceArn` si desea que solo se asocie un recurso al acceso entre servicios. Utiliza `aws:SourceAccount` si quiere permitir que cualquier recurso de esa cuenta se asocie al uso entre servicios. En el caso de MSK Connect, el valor de `aws:SourceArn` debe ser un conector de MSK. La forma más eficaz de protegerse contra el problema de la sustitución confusa es utilizar la clave de contexto de condición global de `aws:SourceArn` con el ARN completo del recurso. Si no conoce el ARN completo del recurso o si especifica varios recursos, utiliza la clave de condición de contexto global `aws:SourceArn` con comodines (`*`) para las partes desconocidas del ARN. Por ejemplo, *arn:aws:kafkaconnect:us-east-1:123456789012:connector/\$1* representa todos los conectores que pertenecen a la cuenta con el identificador 123456789012 en la región EE.UU. Este (Norte de Virginia). El siguiente ejemplo muestra cómo se pueden utilizar las claves contextuales de condición global `aws:SourceArn` y `aws:SourceAccount` en MSK Connect para evitar el problema del adjunto confundido. Sustituya *123456789012* y arn:aws:kafkaconnect: ::connector//por la información de su conector y. *us-east-1* *123456789012* *my-S3-Sink-Connector* *abcd1234-5678-90ab-cdef-1234567890ab* Cuenta de AWS ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": " kafkaconnect.amazonaws.com" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "aws:SourceAccount": "123456789012" }, "ArnLike": { "aws:SourceArn": "arn:aws:kafkaconnect:us-east-1:123456789012:connector/my-S3-Sink-Connector/abcd1234-5678-90ab-cdef-1234567890ab" } } } ] } ``` ------ # AWS políticas administradas para MSK Connect Una política AWS administrada es una política independiente creada y administrada por. AWS AWS Las políticas administradas están diseñadas para proporcionar permisos para muchos casos de uso comunes, de modo que pueda empezar a asignar permisos a usuarios, grupos y funciones. Ten en cuenta que es posible que las políticas AWS administradas no otorguen permisos con privilegios mínimos para tus casos de uso específicos, ya que están disponibles para que los usen todos los AWS clientes. Se recomienda definir [políticas administradas por el cliente](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#customer-managed-policies) específicas para sus casos de uso a fin de reducir aún más los permisos. No puedes cambiar los permisos definidos en AWS las políticas administradas. Si AWS actualiza los permisos definidos en una política AWS administrada, la actualización afecta a todas las identidades principales (usuarios, grupos y roles) a las que está asociada la política. AWS es más probable que actualice una política AWS administrada cuando Servicio de AWS se lance una nueva o cuando estén disponibles nuevas operaciones de API para los servicios existentes. Para obtener más información, consulte [Políticas administradas por AWS](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#aws-managed-policies) en la *Guía del usuario de IAM*. ## AWS política gestionada: Amazon MSKConnect ReadOnlyAccess Esta política otorga al usuario los permisos necesarios para enumerar y describir los recursos de MSK Connect. Puede asociar la política `AmazonMSKConnectReadOnlyAccess` a las identidades de IAM. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kafkaconnect:ListConnectors", "kafkaconnect:ListCustomPlugins", "kafkaconnect:ListWorkerConfigurations" ], "Resource": "*" }, { "Effect": "Allow", "Action": [ "kafkaconnect:DescribeConnector" ], "Resource": [ "arn:aws:kafkaconnect:*:*:connector/*" ] }, { "Effect": "Allow", "Action": [ "kafkaconnect:DescribeCustomPlugin" ], "Resource": [ "arn:aws:kafkaconnect:*:*:custom-plugin/*" ] }, { "Effect": "Allow", "Action": [ "kafkaconnect:DescribeWorkerConfiguration" ], "Resource": [ "arn:aws:kafkaconnect:*:*:worker-configuration/*" ] } ] } ``` ------ ## AWS política gestionada: KafkaConnectServiceRolePolicy Esta política otorga al servicio MSK Connect los permisos necesarios para crear y administrar las interfaces de red que tienen la etiqueta `AmazonMSKConnectManaged:true`. Estas interfaces de red proporcionan acceso de red de MSK Connect a los recursos de su Amazon VPC, como un clúster de Apache Kafka o un origen o un receptor. No puede adjuntarse KafkaConnectServiceRolePolicy a sus entidades de IAM. Esta política está adjunta a un rol vinculado a servicios que permite a MSK Connect realizar acciones en su nombre. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "ec2:CreateNetworkInterface" ], "Resource": "arn:aws:ec2:*:*:network-interface/*", "Condition": { "StringEquals": { "aws:RequestTag/AmazonMSKConnectManaged": "true" }, "ForAllValues:StringEquals": { "aws:TagKeys": "AmazonMSKConnectManaged" } } }, { "Effect": "Allow", "Action": [ "ec2:CreateNetworkInterface" ], "Resource": [ "arn:aws:ec2:*:*:subnet/*", "arn:aws:ec2:*:*:security-group/*" ] }, { "Effect": "Allow", "Action": [ "ec2:CreateTags" ], "Resource": "arn:aws:ec2:*:*:network-interface/*", "Condition": { "StringEquals": { "ec2:CreateAction": "CreateNetworkInterface" } } }, { "Effect": "Allow", "Action": [ "ec2:DescribeNetworkInterfaces", "ec2:CreateNetworkInterfacePermission", "ec2:AttachNetworkInterface", "ec2:DetachNetworkInterface", "ec2:DeleteNetworkInterface" ], "Resource": "arn:aws:ec2:*:*:network-interface/*", "Condition": { "StringEquals": { "ec2:ResourceTag/AmazonMSKConnectManaged": "true" } } } ] } ``` ------ ## MSK Connect actualiza las políticas AWS gestionadas Consulta los detalles sobre las actualizaciones de las políticas AWS administradas de MSK Connect desde que este servicio comenzó a rastrear estos cambios. | Cambio | Descripción | Fecha | | --- | --- | --- | | MSK Connect actualizó la política de solo lectura | MSK Connect actualizó la MSKConnect ReadOnlyAccess política de Amazon para eliminar las restricciones a las operaciones de publicación de anuncios. | 13 de octubre de 2021 | | MSK Connect comenzó a realizar un seguimiento de los cambios | MSK Connect comenzó a realizar un seguimiento de los cambios en sus políticas AWS gestionadas. | 14 de septiembre de 2021 | # Uso de roles vinculados a servicios para MSK Connect Amazon MSK Connect utiliza funciones vinculadas a [servicios AWS Identity and Access Management](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_terms-and-concepts.html#iam-term-service-linked-role) (IAM). Un rol vinculado a servicios es un tipo único de rol de IAM que está vinculado directamente a MSK Connect. MSK Connect predefine las funciones vinculadas al servicio e incluyen todos los permisos que el servicio requiere para llamar a AWS otros servicios en su nombre. Un rol vinculado a un servicio simplifica la configuración de MSK Connect porque ya no tendrá que agregar manualmente los permisos necesarios. MSK Connect define los permisos de sus roles vinculados a servicios y, a menos que esté definido de otra manera, solo MSK Connect puede asumir sus roles. Los permisos definidos incluyen las políticas de confianza y de permisos, y que la política de permisos no se pueda asociar a ninguna otra entidad de IAM. Para obtener información acerca de otros servicios que admiten roles vinculados a servicios, consulte [Servicios de AWS que funcionan con IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_aws-services-that-work-with-iam.html) y busque los servicios que muestran **Yes **(Sí) en la columna **Service Linked Role** (Rol vinculado a servicios). Elija una opción **Sí** con un enlace para ver la documentación acerca del rol vinculado al servicio en cuestión. ## Permisos de roles vinculados a servicios de MSK Connect MSK Connect utiliza el rol vinculado al servicio denominado — **AWSServiceRoleForKafkaConnect**Permite que Amazon MSK Connect acceda a los recursos de Amazon en su nombre. El rol AWSService RoleForKafkaConnect vinculado al servicio confía en que el servicio asuma el `kafkaconnect.amazonaws.com` rol. Para obtener información sobre la política de permisos que utiliza el rol, consulte [AWS política gestionada: KafkaConnectServiceRolePolicy](mkc-security-iam-awsmanpol.md#security-iam-awsmanpol-KafkaConnectServiceRolePolicy). Debe configurar permisos para permitir a una entidad de IAM (como un usuario, grupo o rol) crear, editar o eliminar un rol vinculado a servicios. Para obtener más información, consulte [Permisos de roles vinculados a servicios](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#service-linked-role-permissions) en la *Guía del usuario de IAM*. ## Creación de un rol vinculado al servicio para MSK Connect No necesita crear manualmente un rol vinculado a servicios. Al crear un conector en la Consola de administración de AWS, la o la AWS API AWS CLI, MSK Connect crea automáticamente la función vinculada al servicio. Si elimina este rol vinculado a servicios y necesita crearlo de nuevo, puede utilizar el mismo proceso para volver a crear el rol en su cuenta. Al crear un conector, MSK Connect se encarga de crear de nuevo el rol vinculado al servicio por usted. ## Modificación de un rol vinculado al servicio para MSK Connect MSK Connect no le permite editar el rol vinculado al AWSService RoleForKafkaConnect servicio. Después de crear un rol vinculado a un servicio, no puede cambiarle el nombre, ya que varias entidades pueden hacer referencia a él. Sin embargo, puede editar la descripción del rol mediante IAM. Para obtener más información, consulte [Edición de un rol vinculado a servicios](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#edit-service-linked-role) en la *Guía del usuario de IAM*. ## Eliminación de un rol vinculado a un servicio en MSK Connect Puede utilizar la consola de IAM AWS CLI o la AWS API para eliminar manualmente el rol vinculado al servicio. Para ello, primero elimine manualmente todos los conectores de MSK Connect y, a continuación, elimine manualmente el rol. Para obtener más información, consulte [Eliminación de un rol vinculado a servicios](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#delete-service-linked-role) en la *Guía del usuario de IAM*. ## Regiones admitidas para los roles vinculados al servicio de MSK Connect MSK Connect admite el uso de roles vinculados al servicio en todas las regiones en las que el servicio esté disponible. Para obtener más información, consulte [Regiones y puntos de conexión de AWS](https://docs.aws.amazon.com/general/latest/gr/rande.html). # Habilitación del acceso a Internet para Amazon MSK Connect Si su conector para Amazon MSK Connect necesita acceso a Internet, le recomendamos que utilice la siguiente configuración Amazon Virtual Private Cloud (VPC) para habilitar ese acceso. + Configure el conector con subredes privadas. + Cree una [puerta de enlace NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) pública o una [instancia de NAT](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_NAT_Instance.html) para su VPC en una subred pública. Para obtener más información, consulte la página [Conectar subredes a Internet u otros dispositivos VPCs mediante NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html) en la *Guía del *Amazon Virtual Private Cloud*usuario*. + Permita el tráfico saliente de sus subredes privadas a su instancia o puerta de enlace NAT. # Configuración de una puerta de enlace NAT para Amazon MSK Connect Los pasos siguientes muestran cómo configurar una puerta de enlace NAT para habilitar el acceso a Internet de un conector. Debe completar estos pasos antes de crear un conector en una subred privada. ## Complete los requisitos previos para configurar una puerta de enlace NAT Asegúrese de tener los siguientes elementos. + El ID de la Amazon Virtual Private Cloud (VPC) asociada al clúster. Por ejemplo, *vpc-123456ab*. + Las subredes privadas IDs de su VPC. Por ejemplo, *subnet-a1b2c3de*, *subnet-f4g5h6ij*, etc. Debe configurar el conector con subredes privadas. ## Pasos para habilitar el acceso a Internet para su conector **Habilitación del acceso a Internet para su conector** 1. Abra la Amazon Virtual Private Cloud consola en. [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/) 1. Cree una subred pública para su puerta de enlace NAT con un nombre descriptivo y anote el ID de subred. Para obtener instrucciones detalladas, consulte [Crear una subred en la VPC](https://docs.aws.amazon.com/vpc/latest/userguide/working-with-vpcs.html#AddaSubnet). 1. Cree una puerta de enlace de Internet para que su VPC pueda comunicarse con Internet y anote el ID de la puerta de enlace. Adjunte una puerta de enlace de Internet a su VPC. Para obtener más instrucciones, consulte [Crear y adjuntar una puerta de enlace de Internet](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Attach_Gateway). 1. Aprovisione una puerta de enlace NAT pública para que los hosts de sus subredes privadas puedan acceder a su subred pública. Cuando cree la puerta de enlace NAT, seleccione la subred pública que creó anteriormente. Para obtener instrucciones, consulte [Creación de una puerta de enlace NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html#nat-gateway-creating). 1. Configure sus tablas de enrutamiento. Debe tener dos tablas de enrutamiento en total para completar esta configuración. Ya debería tener una tabla de enrutamiento principal que se haya creado automáticamente al mismo tiempo que su VPC. En este paso, crea una tabla de enrutamiento adicional para su subred pública. 1. Utilice la siguiente configuración para modificar la tabla de enrutamiento principal de la VPC de modo que las subredes privadas dirijan el tráfico a la puerta de enlace NAT. Para obtener instrucciones, consulte [Trabajar con tablas de enrutamiento](https://docs.aws.amazon.com/vpc/latest/userguide/WorkWithRouteTables.html) en la *Guía del usuario* de *Amazon Virtual Private Cloud*. **Tabla de enrutamiento Private MSKC** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/msk/latest/developerguide/msk-connect-internet-access-private-subnets-example.html) 1. Siga las instrucciones en [Crear de una tabla de enrutamiento personalizada](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Routing) para crear una para la subred pública. Al crear la tabla, introduzca un nombre descriptivo en el campo **Etiqueta de nombre** para ayudarle a identificar la subred a la que está asociada la tabla. Por ejemplo, **Public MSKC**. 1. Configure su tabla de enrutamiento **Public MSKC** con los siguientes ajustes. **** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/msk/latest/developerguide/msk-connect-internet-access-private-subnets-example.html) # Descripción de los nombres de host de DNS privados Con la compatibilidad con nombres de host DNS privados en MSK Connect, puede configurar conectores para que hagan referencia a nombres de dominio públicos o privados. El soporte depende de los servidores DNS especificados en el *conjunto de opciones de DHCP* de la VPC. Un conjunto de opciones de DHCP es un grupo de configuraciones de red que las instancias de EC2 en la VPC utilizan para comunicarse a través de la red VPC. Cada VPC dispone de un conjunto de opciones de DHCP predeterminado, pero puede crear un conjunto de opciones de DHCP personalizado si, por ejemplo, desea que las instancias de la VPC utilicen un servidor DNS diferente para la resolución de nombres de dominio en lugar del servidor DNS de Amazon. Consulte [Conjuntos de opciones de DHCP en Amazon VPC](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_DHCP_Options.html). Antes de incluir la resolución de DNS privado en MSK Connect, los conectores capability/feature utilizaban los solucionadores de DNS de VPC del servicio para las consultas de DNS desde un conector de cliente. Los conectores no utilizaban los servidores DNS definidos en los conjuntos de opciones de DHCP de la VPC del cliente para la resolución de DNS. Los conectores solo podían hacer referencia a los nombres de host en las configuraciones de los conectores del cliente o en los complementos que se pudieran resolver públicamente. No podían resolver los nombres de host privados definidos en una zona alojada de forma privada ni utilizar servidores DNS en la red de otro cliente. Sin el DNS privado, los clientes que optaran por hacer que sus bases de datos, almacenes de datos y sistemas, como el Secrets Manager de su propia VPC, fueran inaccesibles a Internet, no podían trabajar con los conectores de MSK. Los clientes suelen utilizar nombres de servidor DNS privados para cumplir con las normas de seguridad corporativas. # Configuración de un conjunto de opciones de DHCP de la VPC para su conector Los conectores utilizan automáticamente los servidores DNS definidos en su conjunto de opciones de DHCP de VPC al crear el conector. Antes de crear un conector, asegúrese de configurar el conjunto de opciones de DHCP de VPC para los requisitos de resolución de nombres de host DNS del conector. Los conectores que creó antes de que la característica de nombre de host de DNS privado estuviera disponible en MSK Connect siguen utilizando la configuración de resolución de DNS anterior sin necesidad de modificarlos. Si solo necesita una resolución de nombres de host DNS de resolución pública en su conector, para facilitar la configuración, le recomendamos que utilice la VPC predeterminada de su cuenta al crear el conector. Consulte [Servidor DNS de Amazon](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#AmazonDNS) en la *Guía del usuario de Amazon VPC* para obtener más información sobre el servidor DNS proporcionado por Amazon o Amazon Route 53 Resolver. Si necesita resolver los nombres de host DNS privados, asegúrese de que la VPC que se transfiere durante la creación del conector tenga el conjunto de opciones de DHCP configurado correctamente. Para obtener más información, consulte [Trabajar con los conjuntos de opciones de DHCP](https://docs.aws.amazon.com/vpc/latest/userguide/DHCPOptionSet.html) en la *Guía del usuario de Amazon VPC*. Al configurar un conjunto de opciones de DHCP para la resolución de nombres de host de DNS privados, asegúrese de que el conector pueda acceder a los servidores DNS personalizados que configure en el conjunto de opciones de DHCP. De lo contrario, se producirá un error al crear el conector. Tras personalizar el conjunto de opciones de DHCP de la VPC, los conectores que se creen posteriormente en esa VPC utilizan los servidores DNS que especificó en el conjunto de opciones. Si cambia el conjunto de opciones después de crear un conector, el conector adopta la configuración del nuevo conjunto de opciones en un par de minutos. # Configuración de atributos de DNS para su VPC Asegúrese de tener los atributos de DNS de la VPC correctamente configurados, tal y como se describe en la sección [Atributos de DNS de la VPC](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#vpc-dns-support) y los [nombres de host de DNS](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-dns.html#vpc-dns-hostnames) de la *Guía del usuario de Amazon VPC.* Consulte [Resolución de consultas de DNS entre VPCs y su red](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/resolver.html) en la *Guía para desarrolladores de Amazon Route 53* para obtener información sobre el uso de puntos finales de resolución entrantes y salientes para conectar otras redes a su VPC y que funcionen con su conector. # Solución de errores en la creación de conectores Esta sección describe los posibles errores en la creación de conectores relacionados con la resolución de DNS y sugerencias de acciones para resolver los problemas. | Failure | Acción sugerida | | --- | --- | | La creación del conector falla si se produce un error en una consulta de resolución de DNS o si no se puede acceder a los servidores DNS desde el conector. | Si ha configurado estos registros para su conector, puede ver errores en la creación de conectores debido a consultas de resolución de DNS erróneas en sus CloudWatch registros. Compruebe las configuraciones del servidor DNS y asegúrese de que la red esté conectada a los servidores DNS desde el conector. | | Si cambia la configuración de los servidores DNS en el conjunto de opciones de DHCP de la VPC mientras se ejecuta un conector, las consultas de resolución de DNS del conector pueden fallar. Si se produce un error en la resolución de DNS, algunas de las tareas del conector pueden entrar en un estado fallido. | Si has configurado estos CloudWatch registros para tu conector, puedes ver errores en la creación de conectores debido a consultas de resolución de DNS erróneas en tus registros. Las tareas fallidas deberían reiniciarse automáticamente para que el conector vuelva a funcionar. Si eso no ocurre, puede ponerse en contacto con el servicio de asistencia para reiniciar las tareas fallidas en el conector correspondiente o puede volver a crear el conector. | # Seguridad para MSK Connect Puede utilizar un punto de enlace de interfaz VPC, con tecnología de AWS PrivateLink, para evitar que el tráfico entre su Amazon VPC y Amazon MSK-Connect APIs compatible salga de la red de Amazon. Los puntos finales de la interfaz VPC no requieren una puerta de enlace a Internet, un dispositivo NAT, una conexión VPN o una conexión. AWS Direct Connect Para obtener más información, consulte [Utilice Amazon MSK APIs con puntos de enlace de VPC de interfaz](privatelink-vpc-endpoints.md). # Registro de MSK Connect MSK Connect puede escribir eventos de registro que puede usar para depurar el conector. Al crear un conector, puede especificar ninguno o varios de los siguientes destinos de registro: + Amazon CloudWatch Logs: usted especifica el grupo de registros al que quiere que MSK Connect envíe los eventos de registro de su conector. Para obtener información sobre cómo crear un grupo de registros, consulte [Crear un](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Working-with-log-groups-and-streams.html#Create-Log-Group) grupo de *CloudWatch registros en la Guía del usuario* de Logs. + Amazon S3: usted especifica el bucket de S3 al que quiere que MSK Connect envíe los eventos de registro de su conector. Para obtener más información sobre la creación de un bucket de S3, consulte [Crear un bucket](https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket-overview.html) en la *Guía del usuario de Amazon S3*. + Amazon Data Firehose: especifica el flujo de entrega al que quiere que MSK Connect envíe los eventos de registro de su conector. Para obtener información sobre la creación de un flujo de entrega, consulte [Creación de un flujo de entrega de Amazon Data Firehose](https://docs.aws.amazon.com/firehose/latest/dev/basic-create.html) en la *Guía del usuario de Firehose*. Para obtener más información sobre la configuración del registro, consulte [Habilitar el registro desde determinados servicios de AWS](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/AWS-logs-and-resource-policy.html) en la *Guía del usuario de Amazon CloudWatch Logs *. MSK Connect emite los siguientes tipos de eventos de registro: **** | Nivel | Description (Descripción) | | --- | --- | | INFO | Eventos de tiempo de ejecución de interés durante el inicio y el cierre. | | WARN | Situaciones de tiempo de ejecución que no son errores pero que son indeseables o inesperadas. | | FATAL | Errores graves que provocan una terminación prematura. | | ERROR | Condiciones inesperadas y errores de tiempo de ejecución que no son irrecuperables. | A continuación, se muestra un ejemplo de un evento de registro enviado a CloudWatch Logs: ``` [Worker-0bb8afa0b01391c41] [2021-09-06 16:02:54,151] WARN [Producer clientId=producer-1] Connection to node 1 (b-1.my-test-cluster.twwhtj.c2.kafka.us-east-1.amazonaws.com/INTERNAL_IP) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:782) ``` ## Impedir que los secretos aparezcan en los registros de los conectores **nota** Los valores de configuración confidenciales pueden aparecer en los registros de los conectores si un complemento no los define como secretos. Kafka Connect trata los valores de configuración indefinidos de la misma manera que cualquier otro valor de texto sin formato. Si su complemento define una propiedad como secreta, Kafka Connect redacta el valor de la propiedad de los registros del conector. Por ejemplo, los siguientes registros de conectores demuestran que si un complemento define `aws.secret.key` como un tipo `PASSWORD`, su valor se sustituye por `[hidden]`. ``` 2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] [2022-01-11 15:18:55,150] INFO SecretsManagerConfigProviderConfig values: 2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] aws.access.key = my_access_key 2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] aws.region = us-east-1 2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] aws.secret.key = [hidden] 2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] secret.prefix = 2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] secret.ttl.ms = 300000 2022-01-11T15:18:55.000+00:00 [Worker-05e6586a48b5f331b] (com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProviderConfig:361) ``` Para evitar que los secretos aparezcan en los archivos de registro de los conectores, el desarrollador de un complemento debe usar la constante de enumeración de Kafka Connect [https://kafka.apache.org/27/javadoc/org/apache/kafka/common/config/ConfigDef.Type.html#PASSWORD](https://kafka.apache.org/27/javadoc/org/apache/kafka/common/config/ConfigDef.Type.html#PASSWORD) para definir las propiedades confidenciales. Cuando una propiedad es del tipo `ConfigDef.Type.PASSWORD`, Kafka Connect excluye su valor de los registros del conector incluso si el valor se envía como texto sin formato. # Supervisión de Amazon MSK Connect La supervisión es una parte importante del mantenimiento de la fiabilidad, la disponibilidad y el rendimiento de MSK Connect y sus demás AWS soluciones. Amazon CloudWatch monitorea tus AWS recursos y las aplicaciones en las que ejecutas AWS en tiempo real. Puede recopilar métricas y realizar un seguimiento de las métricas, crear paneles personalizados y definir alarmas que le advierten o que toman medidas cuando una métrica determinada alcanza el umbral que se especifique. Por ejemplo, puede CloudWatch hacer un seguimiento del uso de la CPU u otras métricas de su conector, de modo que pueda aumentar su capacidad si es necesario. Para obtener más información, consulta la [Guía del CloudWatch usuario de Amazon](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/). Puede usar las siguientes operaciones de la API: + `DescribeConnectorOperation`: supervise el estado de las operaciones de actualización del conector. + `ListConnectorOperations`: realice un seguimiento de las actualizaciones anteriores ejecutadas en el conector. En la siguiente tabla, se muestran las métricas a las que MSK Connect envía CloudWatch en la `ConnectorName` dimensión. MSK Connect proporciona estas métricas de forma predeterminada y sin coste adicional. CloudWatch conserva estas métricas durante 15 meses, para que pueda acceder a la información histórica y obtener una mejor perspectiva del rendimiento de sus conectores. También puede establecer alarmas que vigilen determinados umbrales y enviar notificaciones o realizar acciones cuando se cumplan dichos umbrales. Para obtener más información, consulta la [Guía del CloudWatch usuario de Amazon](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/). | Nombre de métrica | Description (Descripción) | | --- | --- | | CpuUtilization | El porcentaje de consumo de CPU por sistema y usuario. | | ErroredTaskCount | El número de tareas que han producido errores. | | MemoryUtilization | El porcentaje de la memoria total de una instancia de proceso de trabajo, no solo de la memoria en montón de la máquina virtual Java (JVM) que se utiliza actualmente. Por lo general, la JVM no devuelve memoria al sistema operativo. Por lo tanto, el tamaño de pila de JVM (MemoryUtilization) normalmente comienza con un tamaño de pila mínimo que aumenta gradualmente hasta un máximo estable de aproximadamente el 80-90%. El uso en montón de JVM puede aumentar o disminuir a medida que cambia el uso real de la memoria del conector. | | RebalanceCompletedTotal | El número total de reequilibrados realizados por este conector. | | RebalanceTimeAvg | El tiempo medio en milisegundos que tarda el conector en reequilibrarse. | | RebalanceTimeMax | El tiempo máximo en milisegundos que tarda el conector en reequilibrarse. | | RebalanceTimeSinceLast | El tiempo en milisegundos desde que este conector completó el reequilibrio más reciente. | | RunningTaskCount | El número de tareas en ejecución en el conector. | | SinkConsumerByteRate | El número promedio de bytes consumidos por segundo por el consumidor de tipo Sink del marco Kafka Connect antes de que se apliquen transformaciones a los datos. | | SinkRecordReadRate | El número medio de registros leídos por segundo desde el clúster de Apache Kafka o Amazon MSK. | | SinkRecordSendRate | El número medio por segundo de registros que se generan a partir de las transformaciones y se envían al destino. Este número no incluye los registros filtrados. | | SourceRecordPollRate | El número medio por segundo de registros producidos o sondeados. | | SourceProducerByteRate | El número promedio de bytes producidos por segundo por el productor de tipo Source del marco Kafka Connect después de que se apliquen transformaciones a los datos. | | SourceRecordWriteRate | El número medio de salida de registros por segundo desde las transformaciones y escrituras en el clúster de Apache Kafka o Amazon MSK. | | TaskStartupAttemptsTotal | El número total de intentos de inicio de tareas del conector. Puede usar esta métrica para identificar anomalías en los intentos de inicio de tareas. | | TaskStartupSuccessPercentage | El porcentaje medio de inicios de tareas satisfactorios del conector. Puede usar esta métrica para identificar anomalías en los intentos de inicio de tareas. | | WorkerCount | El número de procesos de trabajo que se están ejecutando en el conector. | | BytesInPerSec | Bytes de metadatos transferidos al marco de trabajo Kafka Connect para la comunicación entre nodos de trabajo. | | BytesOutPerSec | Bytes de metadatos transferidos desde el marco de trabajo Kafka Connect para la comunicación entre nodos de trabajo. | # Ejemplos de configuración de los recursos de Amazon MSK Connect En esta sección se incluyen ejemplos que le ayudarán a configurar los recursos de Amazon MSK Connect, como los conectores y proveedores de configuración habituales de terceros. **Topics** + [Configuración de un conector de recepción de Amazon S3](mkc-S3sink-connector-example.md) + [Configurar el conector de colector EventBridge Kafka para MSK Connect](mkc-eventbridge-kafka-connector.md) + [Uso del conector de origen Debezium con un proveedor de configuración](mkc-debeziumsource-connector-example.md) # Configuración de un conector de recepción de Amazon S3 En este ejemplo se muestra cómo utilizar el [conector receptor Amazon S3 de](https://www.confluent.io/hub/confluentinc/kafka-connect-s3) Confluent y cómo AWS CLI crear un conector receptor Amazon S3 en MSK Connect. 1. Copie el siguiente JSON y péguelo en un nuevo archivo. Sustituya las cadenas de marcador de posición por valores que correspondan a la cadena de conexión de los servidores de arranque del clúster de Amazon MSK y a la subred y al grupo de seguridad del clúster. IDs Para obtener más información sobre cómo configurar un rol de ejecución del servicio, consulte [Políticas y roles de IAM para MSK Connect](msk-connect-iam.md). ``` { "connectorConfiguration": { "connector.class": "io.confluent.connect.s3.S3SinkConnector", "s3.region": "us-east-1", "format.class": "io.confluent.connect.s3.format.json.JsonFormat", "flush.size": "1", "schema.compatibility": "NONE", "topics": "my-test-topic", "tasks.max": "2", "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner", "storage.class": "io.confluent.connect.s3.storage.S3Storage", "s3.bucket.name": "amzn-s3-demo-bucket" }, "connectorName": "example-S3-sink-connector", "kafkaCluster": { "apacheKafkaCluster": { "bootstrapServers": "", "vpc": { "subnets": [ "", "", "" ], "securityGroups": [""] } } }, "capacity": { "provisionedCapacity": { "mcuCount": 2, "workerCount": 4 } }, "kafkaConnectVersion": "2.7.1", "serviceExecutionRoleArn": "", "plugins": [ { "customPlugin": { "customPluginArn": "", "revision": 1 } } ], "kafkaClusterEncryptionInTransit": {"encryptionType": "PLAINTEXT"}, "kafkaClusterClientAuthentication": {"authenticationType": "NONE"} } ``` 1. Ejecute el siguiente AWS CLI comando en la carpeta en la que guardó el archivo JSON en el paso anterior. ``` aws kafkaconnect create-connector --cli-input-json file://connector-info.json ``` El siguiente es un ejemplo del resultado que se obtiene al ejecutar el comando correctamente. ``` { "ConnectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-S3-sink-connector/abc12345-abcd-4444-a8b9-123456f513ed-2", "ConnectorState": "CREATING", "ConnectorName": "example-S3-sink-connector" } ``` # Configurar el conector de colector EventBridge Kafka para MSK Connect En este tema se muestra cómo configurar el [conector colector EventBridge Kafka](https://github.com/awslabs/eventbridge-kafka-connector) para MSK Connect. [Este conector le permite enviar eventos desde su clúster de MSK a EventBridge los buses de eventos.](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-bus.html) En este tema se describe el proceso para crear los recursos necesarios y configurar el conector para permitir un flujo de datos fluido entre Kafka y. EventBridge **Topics** + [Requisitos previos](#mkc-eb-kafka-prerequisites) + [Configuración de los recursos necesarios para MSK Connect](#mkc-eb-kafka-set-up-resources) + [Creación del conector](#mkc-eb-kafka-create-connector) + [Envío de mensajes a Kafka](#mkc-eb-kafka-send-json-encoded-messages) ## 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. + **Amazon EventBridge Event Bus**: un autobús de EventBridge eventos para recibir eventos relacionados con los temas de Kafka. + Funciones de **IAM: cree funciones** de IAM con los permisos necesarios para MSK Connect y EventBridge el conector. + [Acceso a Internet público](msk-connect-internet-access.md) desde MSK Connect o desde un punto final EventBridge de [interfaz de VPC](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-related-service-vpc.html) creado en la VPC y la subred de su 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](create-serverless-cluster-client.md), como una instancia de Amazon EC2 o [AWS CloudShell](https://aws.amazon.com/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 debe crear una EventBridge regla para filtrar los eventos de Kafka enviados al bus de eventos. EventBridge **Topics** + [Rol de IAM para el conector](#mkc-eb-kafka-iam-role-connector) + [Una EventBridge regla para los eventos entrantes](#mkc-eb-kafka-create-rule) ### Rol de IAM para el conector La función de IAM que asocie al conector debe tener el [PutEvents](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-permissions-reference.html)permiso para permitir el envío de eventos. 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 EventBridge regla para los eventos entrantes Crea [reglas](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-rules.html) que coinciden con los eventos entrantes en función de criterios de datos de eventos, conocidos como [https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html). 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](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-targets.html) designado. El siguiente ejemplo de un patrón de eventos coincide con los eventos de Kafka enviados al bus de EventBridge eventos. ``` { "detail": { "topic": ["msk-eventbridge-tutorial"] } } ``` A continuación se muestra un ejemplo de un evento enviado desde Kafka para EventBridge utilizar el conector colector 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 EventBridge consola, [cree una regla](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-create-rule.html) en el bus de eventos con este patrón de ejemplo y especifique un objetivo, como un grupo de CloudWatch registros. La EventBridge consola configurará automáticamente la política de acceso necesaria para el grupo de CloudWatch registros. ## Creación del conector En la siguiente sección, creará e implementará el [conector receptor de EventBridge Kafka](https://github.com/awslabs/eventbridge-kafka-connector) mediante el Consola de administración de AWS. **Topics** + [Paso 1: Descarga del conector](#mkc-eb-kafka-download-connector) + [Paso 2: creación de un bucket de Amazon S3](#mkc-eb-kafka-s3-bucket-create) + [Paso 3: Creación de un complemento en MSK Connect](#mkc-eb-kafka-create-plugin) + [Paso 4: creación del conector](#mkc-eb-kafka-create-connector) ### Paso 1: Descarga del conector Descargue la versión JAR EventBridge del colector de conectores más reciente de la [página de GitHub versiones](https://github.com/awslabs/eventbridge-kafka-connector/releases) del conector EventBridge Kafka. 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 usarlo con MSK Connect, abra y Consola de administración de AWS, a continuación, elija Amazon S3. 1. En la consola de Amazon S3, seleccione **Crear bucket** e introduzca un nombre de bucket único. Por ejemplo, **amzn-s3-demo-bucket1-eb-connector**. 1. 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. 1. En **Configuración del bucket**, mantenga las selecciones predeterminadas o ajústelas según sea necesario. 1. Elija **Crear bucket** 1. Cargue el archivo JAR en el bucket de Amazon S3. ### Paso 3: Creación de un complemento en MSK Connect 1. Abre y Consola de administración de AWS, a continuación, navega hasta **MSK Connect**. 1. En el panel de navegación izquierdo, seleccione **Complementos personalizados**. 1. Seleccione **Crear complemento** y, a continuación, introduzca un **Nombre del complemento**. Por ejemplo, **eventbridge-sink-plugin**. 1. En **Ubicación del complemento personalizado**, pegue la **URL del objeto de S3**. 1. Agregue una descripción opcional para el complemento. 1. Seleccione **Crear complemento**. Una vez creado el complemento, puede usarlo para configurar e implementar el conector EventBridge Kafka 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**. 1. En la lista de complementos, seleccione **eventbridge-sink-plugin** y, a continuación, seleccione **Siguiente**. 1. En nombre del conector, introduzca **EventBridgeSink**. 1. En la lista de clústeres, seleccione el clúster de MSK. 1. 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 solía PrivateLink conectarse de forma segura desde MSK a EventBridge, sustituya la parte de DNS posterior `https://` por el nombre de DNS privado correcto del punto final de la interfaz de VPC (opcional) que EventBridge creó anteriormente. + Sustituya el ARN del bus de EventBridge eventos 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 los conectores, consulte. [eventbridge-kafka-connector](https://github.com/awslabs/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. También recomendamos activar el registro en CloudWatch para facilitar la observación y solucionar 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](#connector-ex) 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. # Uso del conector de origen Debezium con un proveedor de configuración En este ejemplo se muestra cómo utilizar el complemento de conector Debezium MySQL con una base de datos de [Amazon Aurora](https://aws.amazon.com/rds/aurora/) compatible con MySQL como origen. En este ejemplo, también configuramos el proveedor de código abierto [AWS Secrets Manager Config Provider](https://github.com/jcustenborder/kafka-config-provider-aws) para externalizar las credenciales de la base de datos en AWS Secrets Manager. Para obtener más información sobre los proveedores de configuración, consulte [Tutorial: Externalización de la información confidencial con proveedores de configuración](msk-connect-config-provider.md). **importante** El complemento de conector Debezium MySQL [solo admite una tarea](https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-property-tasks-max) y no funciona con el modo de capacidad con escalado automático para Amazon MSK Connect. En su lugar, debe utilizar el modo de capacidad aprovisionada y establecer `workerCount` igual a uno en la configuración del conector. Para obtener más información sobre los modos de capacidad de MSK Connect, consulte [Descripción de la capacidad de los conectores](msk-connect-capacity.md). # Requisitos previos que se deben completar para usar el conector de origen de Debezium Su conector debe poder acceder a Internet para poder interactuar con servicios como los AWS Secrets Manager que están fuera del suyo Amazon Virtual Private Cloud. Los pasos de esta sección le ayudan a realizar las siguientes tareas para habilitar el acceso a Internet. + Configure una subred pública que aloje una puerta de enlace NAT y dirija el tráfico a una puerta de enlace de Internet en su VPC. + Cree una ruta predeterminada que dirija el tráfico de su subred privada a su puerta de enlace NAT. Para obtener más información, consulte [Habilitación del acceso a Internet para Amazon MSK Connect](msk-connect-internet-access.md). **Requisitos previos** Antes de habilitar el acceso a Internet, necesitará lo siguiente: + El ID de la Amazon Virtual Private Cloud (VPC) asociada al clúster. Por ejemplo, *vpc-123456ab*. + Las subredes privadas IDs de su VPC. Por ejemplo, *subnet-a1b2c3de*, *subnet-f4g5h6ij*, etc. Debe configurar el conector con subredes privadas. **Habilitación del acceso a Internet para su conector** 1. Abra la Amazon Virtual Private Cloud consola en. [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/) 1. Cree una subred pública para su puerta de enlace NAT con un nombre descriptivo y anote el ID de subred. Para obtener instrucciones detalladas, consulte [Crear una subred en la VPC](https://docs.aws.amazon.com/vpc/latest/userguide/working-with-vpcs.html#AddaSubnet). 1. Cree una puerta de enlace de Internet para que su VPC pueda comunicarse con Internet y anote el ID de la puerta de enlace. Adjunte una puerta de enlace de Internet a su VPC. Para obtener más instrucciones, consulte [Crear y adjuntar una puerta de enlace de Internet](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Attach_Gateway). 1. Aprovisione una puerta de enlace NAT pública para que los hosts de sus subredes privadas puedan acceder a su subred pública. Cuando cree la puerta de enlace NAT, seleccione la subred pública que creó anteriormente. Para obtener instrucciones, consulte [Creación de una puerta de enlace NAT](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-nat-gateway.html#nat-gateway-creating). 1. Configure sus tablas de enrutamiento. Debe tener dos tablas de enrutamiento en total para completar esta configuración. Ya debería tener una tabla de enrutamiento principal que se haya creado automáticamente al mismo tiempo que su VPC. En este paso, crea una tabla de enrutamiento adicional para su subred pública. 1. Utilice la siguiente configuración para modificar la tabla de enrutamiento principal de la VPC de modo que las subredes privadas dirijan el tráfico a la puerta de enlace NAT. Para obtener instrucciones, consulte [Trabajar con tablas de enrutamiento](https://docs.aws.amazon.com/vpc/latest/userguide/WorkWithRouteTables.html) en la *Guía del usuario* de *Amazon Virtual Private Cloud*. **Tabla de enrutamiento Private MSKC** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html) 1. Siga las instrucciones en [Crear de una tabla de enrutamiento personalizada](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_Internet_Gateway.html#Add_IGW_Routing) para crear una para la subred pública. Al crear la tabla, introduzca un nombre descriptivo en el campo **Etiqueta de nombre** para ayudarle a identificar la subred a la que está asociada la tabla. Por ejemplo, **Public MSKC**. 1. Configure su tabla de enrutamiento **Public MSKC** con los siguientes ajustes. **** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/es_es/msk/latest/developerguide/mkc-debeziumsource-connector-example-prereqs.html) Ahora que ha habilitado el acceso a Internet para Amazon MSK Connect, está listo para crear un conector. # Creación de un conector de origen Debezium En este procedimiento, se describe cómo crear un conector de origen Debezium. 1. **Creación de un complemento personalizado** 1. Descargue la última versión estable del complemento MySQL Connector desde el sitio de [Debezium](https://debezium.io/releases/). Anote la versión de lanzamiento de Debezium que haya descargado (la versión 2.x o la antigua serie 1.x). Más adelante en este procedimiento, creará un conector basado en su versión de Debezium. 1. Descargue y extraiga el [proveedor de configuración de AWS Secrets Manager](https://www.confluent.io/hub/jcustenborder/kafka-config-provider-aws). 1. Coloque los siguientes archivos en el mismo directorio: + La carpeta `debezium-connector-mysql` + La carpeta `jcusten-border-kafka-config-provider-aws-0.1.1` 1. Comprima el directorio que creó en el paso anterior en un archivo ZIP y, a continuación, cargue el archivo ZIP en un bucket de S3. Para obtener instrucciones, consulte [Carga de objetos](https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html) en la *Guía del usuario de Amazon S3*. 1. Copie el siguiente JSON y péguelo en un archivo. Por ejemplo, `debezium-source-custom-plugin.json`. **Sustitúyalo por el nombre que desee que tenga el complemento, ** por el ARN del bucket de Amazon S3 en el que cargó el archivo ZIP y `` por la clave de archivo del objeto ZIP que cargó en S3. ``` { "name": "", "contentType": "ZIP", "location": { "s3Location": { "bucketArn": "", "fileKey": "" } } } ``` 1. Ejecute el siguiente AWS CLI comando desde la carpeta en la que guardó el archivo JSON para crear un complemento. ``` aws kafkaconnect create-custom-plugin --cli-input-json file:// ``` Debería ver un resultado similar a este. ``` { "CustomPluginArn": "arn:aws:kafkaconnect:us-east-1:012345678901:custom-plugin/example-custom-plugin-name/abcd1234-a0b0-1234-c1-12345678abcd-1", "CustomPluginState": "CREATING", "Name": "example-custom-plugin-name", "Revision": 1 } ``` 1. Ejecute el siguiente comando para comprobar el estado del complemento. El estado debería cambiar de `CREATING` a `ACTIVE`. Sustituya el marcador de posición del ARN por el ARN que obtuvo en el resultado del comando anterior. ``` aws kafkaconnect describe-custom-plugin --custom-plugin-arn "" ``` 1. **Configura AWS Secrets Manager y crea un secreto para las credenciales de tu base de datos** 1. Abra la consola de Secrets Manager en [https://console.aws.amazon.com/secretsmanager/](https://console.aws.amazon.com/secretsmanager/). 1. Cree un nuevo secreto para almacenar sus credenciales de inicio de sesión de base de datos. Para obtener instrucciones, consulte [Creación de un secreto](https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_create-basic-secret.html) en la Guía del usuario de *AWS Secrets Manager*. 1. Copie el ARN de su secreto. 1. Agregue los permisos de Secrets Manager desde la siguiente política de ejemplo a su [Descripción del rol de ejecución de servicios](msk-connect-service-execution-role.md). **Sustitúyalo por el ARN de su secreto. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "secretsmanager:GetResourcePolicy", "secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret", "secretsmanager:ListSecretVersionIds" ], "Resource": [ "arn:aws:secretsmanager:us-east-1:123456789012:secret:MySecret-1234" ] } ] } ``` ------ Para obtener información sobre cómo administrar los permisos de IAM, consulte [Adición y eliminación de permisos de identidad de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) en la *Guía del usuario de IAM*. 1. **Creación de una configuración de proceso de trabajo personalizada con información sobre su proveedor de configuración** 1. Copie las siguientes propiedades de configuración de proceso de trabajo en un archivo y sustituya las cadenas de marcadores de posición por valores que correspondan a su caso de uso. Para obtener más información sobre las propiedades de configuración del proveedor de configuración de AWS Secrets Manager, consulte [SecretsManagerConfigProvider](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-config-provider-aws/configProviders/SecretsManagerConfigProvider.html)la documentación del complemento. ``` key.converter= value.converter= config.providers.secretManager.class=com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider config.providers=secretManager config.providers.secretManager.param.aws.region= ``` 1. Ejecute el siguiente AWS CLI comando para crear su configuración de trabajo personalizada. Reemplace los siguientes valores: + **- un nombre descriptivo para su configuración de trabajo personalizada + **- una versión codificada en base64 de las propiedades de texto sin formato que copió en el paso anterior ``` aws kafkaconnect create-worker-configuration --name --properties-file-content ``` 1. **Creación de un conector** 1. Copie el siguiente JSON que corresponda a su versión de Debezium (2.x o 1.x) y péguelo en un archivo nuevo. Sustituya las cadenas `` por valores que correspondan a su caso de uso. Para obtener más información sobre cómo configurar un rol de ejecución del servicio, consulte [Políticas y roles de IAM para MSK Connect](msk-connect-iam.md). Tenga en cuenta que la configuración utiliza variables como `${secretManager:MySecret-1234:dbusername}` en lugar de texto sin formato para especificar las credenciales de la base de datos. Sustituya `MySecret-1234` por el nombre de su secreto y, a continuación, incluya el nombre de la clave que desea recuperar. También debe reemplazar `` por el ARN de su configuración de proceso de trabajo personalizada. ------ #### [ Debezium 2.x ] Para las versiones 2.x de Debezium, copie el siguiente JSON y péguelo en un archivo nuevo. Sustituya las cadenas ** por valores que correspondan a su caso de uso. ``` { "connectorConfiguration": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "", "database.port": "3306", "database.user": "<${secretManager:MySecret-1234:dbusername}>", "database.password": "<${secretManager:MySecret-1234:dbpassword}>", "database.server.id": "123456", "database.include.list": "", "topic.prefix": "", "schema.history.internal.kafka.topic": "", "schema.history.internal.kafka.bootstrap.servers": "", "schema.history.internal.consumer.security.protocol": "SASL_SSL", "schema.history.internal.consumer.sasl.mechanism": "AWS_MSK_IAM", "schema.history.internal.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "schema.history.internal.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "schema.history.internal.producer.security.protocol": "SASL_SSL", "schema.history.internal.producer.sasl.mechanism": "AWS_MSK_IAM", "schema.history.internal.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "schema.history.internal.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "include.schema.changes": "true" }, "connectorName": "example-Debezium-source-connector", "kafkaCluster": { "apacheKafkaCluster": { "bootstrapServers": "", "vpc": { "subnets": [ "", "", "" ], "securityGroups": [""] } } }, "capacity": { "provisionedCapacity": { "mcuCount": 2, "workerCount": 1 } }, "kafkaConnectVersion": "2.7.1", "serviceExecutionRoleArn": "", "plugins": [{ "customPlugin": { "customPluginArn": "", "revision": 1 } }], "kafkaClusterEncryptionInTransit": { "encryptionType": "TLS" }, "kafkaClusterClientAuthentication": { "authenticationType": "IAM" }, "workerConfiguration": { "workerConfigurationArn": "", "revision": 1 } } ``` ------ #### [ Debezium 1.x ] Para las versiones 1.x de Debezium, copie el siguiente JSON y péguelo en un archivo nuevo. Sustituya las cadenas ** por valores que correspondan a su caso de uso. ``` { "connectorConfiguration": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "", "database.port": "3306", "database.user": "<${secretManager:MySecret-1234:dbusername}>", "database.password": "<${secretManager:MySecret-1234:dbpassword}>", "database.server.id": "123456", "database.server.name": "", "database.include.list": "", "database.history.kafka.topic": "", "database.history.kafka.bootstrap.servers": "", "database.history.consumer.security.protocol": "SASL_SSL", "database.history.consumer.sasl.mechanism": "AWS_MSK_IAM", "database.history.consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "database.history.consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "database.history.producer.security.protocol": "SASL_SSL", "database.history.producer.sasl.mechanism": "AWS_MSK_IAM", "database.history.producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;", "database.history.producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler", "include.schema.changes": "true" }, "connectorName": "example-Debezium-source-connector", "kafkaCluster": { "apacheKafkaCluster": { "bootstrapServers": "", "vpc": { "subnets": [ "", "", "" ], "securityGroups": [""] } } }, "capacity": { "provisionedCapacity": { "mcuCount": 2, "workerCount": 1 } }, "kafkaConnectVersion": "2.7.1", "serviceExecutionRoleArn": "", "plugins": [{ "customPlugin": { "customPluginArn": "", "revision": 1 } }], "kafkaClusterEncryptionInTransit": { "encryptionType": "TLS" }, "kafkaClusterClientAuthentication": { "authenticationType": "IAM" }, "workerConfiguration": { "workerConfigurationArn": "", "revision": 1 } } ``` ------ 1. Ejecute el siguiente AWS CLI comando en la carpeta en la que guardó el archivo JSON en el paso anterior. ``` aws kafkaconnect create-connector --cli-input-json file://connector-info.json ``` El siguiente es un ejemplo del resultado que se obtiene al ejecutar el comando correctamente. ``` { "ConnectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2", "ConnectorState": "CREATING", "ConnectorName": "example-Debezium-source-connector" } ``` # Actualización de la configuración de un conector Debezium Para actualizar la configuración del conector Debezium, siga estos pasos: 1. Copie el siguiente JSON y péguelo en un archivo nuevo. Sustituya las cadenas `` por valores que correspondan a su caso de uso. ``` { "connectorArn": , "connectorConfiguration": , "currentVersion": } ``` 1. Ejecute el siguiente AWS CLI comando en la carpeta en la que guardó el archivo JSON en el paso anterior. ``` aws kafkaconnect update-connector --cli-input-json file://connector-info.json ``` El siguiente ejemplo muestra la salida cuando el comando se ejecuta correctamente. ``` { "connectorArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2", "connectorOperationArn": "arn:aws:kafkaconnect:us-east-1:123450006789:connector-operation/example-Debezium-source-connector/abc12345-abcd-4444-a8b9-123456f513ed-2/41b6ad56-3184-479b-850a-a8bedd5a02f3", "connectorState": "UPDATING" } ``` 1. Ahora puede ejecutar el siguiente comando para supervisar el estado actual de la operación: ``` aws kafkaconnect describe-connector-operation --connector-operation-arn ``` Para ver un ejemplo de conector Debezium con pasos detallados, consulte [Introducing Amazon MSK Connect - Stream Data to and from Your Apache Kafka Clusters Using Managed Connectors](https://aws.amazon.com/blogs/aws/introducing-amazon-msk-connect-stream-data-to-and-from-your-apache-kafka-clusters-using-managed-connectors/). # Migración a Amazon MSK Connect En esta sección, se describe cómo migrar su aplicación del conector de Apache Kafka a Amazon Managed Streaming para Apache Kafka Connect (Amazon MSK Connect). Para obtener más información sobre las ventajas de la migración a Amazon MSK Connect, consulte [Beneficios de usar Amazon MSK Connect](msk-connect.md#msk-connect-benefits). En esta sección, también se abordan los temas de administración de estados que utilizan Kafka Connect y Amazon MSK Connect, así como los procedimientos para migrar los conectores de origen y de recepción. # Descripción de los temas internos que utiliza Kafka Connect Las aplicaciones de Apache Kafka Connect que se ejecutan en modo distribuido almacenan su estado mediante los temas internos en el clúster de Kafka y la pertenencia a grupos. A continuación se enumeran los valores de configuración correspondientes a los temas internos que se utilizan para las aplicaciones de Kafka Connect: + Tema de configuración, especificado mediante `config.storage.topic` En el tema de configuración, Kafka Connect almacena la configuración de todos los conectores y las tareas que iniciaron los usuarios. Cada vez que los usuarios actualizan la configuración de un conector o cuando se requiere una reconfiguración en el conector (por ejemplo, el conector detecta que puede iniciar más tareas), se emite un registro a este tema. La compactación está habilitada para este tema, por lo que siempre mantiene el último estado de cada entidad. + Tema de desplazamientos, especificado mediante `offset.storage.topic` En el tema de desplazamientos, Kafka Connect almacena los desplazamientos de los conectores de origen. Al igual que en el tema de configuración, la compactación está habilitada para el tema de desplazamientos. Este tema se utiliza para escribir las posiciones de origen únicamente para los conectores de origen que producen datos para Kafka desde sistemas externos. Los conectores de recepción, que leen datos de Kafka y los envían a sistemas externos, almacenan los desplazamientos de consumo en grupos de consumidores habituales de Kafka. + Tema de estado, especificado mediante `status.storage.topic` Con el tema de estado, Kafka Connect almacena el estado actual de los conectores y las tareas. Este tema se utiliza como el punto central para los datos que consultan los usuarios de la API de REST. Con este tema, los usuarios pueden consultar cualquier proceso de trabajo y aún recibir el estado de todos los complementos en ejecución. Como en los temas de configuración y desplazamientos, la compactación también está habilitada para el tema de estado. Además de estos temas, Kafka Connect utiliza ampliamente la API de pertenencia a grupos de Kafka. Los grupos se nombran como el conector. Por ejemplo, en el caso de un conector denominado file-sink, el grupo recibe el nombre. connect-file-sink Cada consumidor del grupo proporciona registros para una sola tarea. Estos grupos y sus desplazamientos se pueden recuperar con las herramientas para grupos de consumidores regulares, como `Kafka-consumer-group.sh`. Para cada conector de recepción, el tiempo de ejecución de Connect ejecuta un grupo de consumidores regular que extrae registros desde Kafka. # Administración del estado de las aplicaciones Amazon MSK Connect De forma predeterminada, Amazon MSK Connect crea tres temas distintos para cada conector de Amazon MSK en el clúster de Kafka, con el fin de almacenar la configuración, los desplazamientos y el estado del conector. Los nombres predeterminados de los temas se estructuran de la siguiente manera: + \$1\$1msk\$1connect\$1configs\$1 \$1 *connector-name* *connector-id* + \$1\$1msk\$1connect\$1status\$1 \$1 *connector-name* *connector-id* + \$1\$1msk\$1connect\$1offsets\$1 \$1 *connector-name* *connector-id* **nota** Para proporcionar continuidad de desplazamiento entre los conectores de origen, puede utilizar el tema de almacenamiento de desplazamiento que desee en lugar del tema predeterminado. Especificar un tema de almacenamiento de desplazamiento lo ayuda a realizar tareas como crear un conector de origen que reanude la lectura desde el último desplazamiento de un conector anterior. Para especificar un tema de almacenamiento de desplazamiento, proporcione un valor para la propiedad [https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-manage-connector-offsets](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-manage-connector-offsets) en la configuración del proceso de trabajo Amazon MSK Connect antes de crear un conector. # Migración de los conectores de origen a Amazon MSK Connect Los conectores de origen son aplicaciones de Apache Kafka Connect que importan registros desde sistemas externos a Kafka. En esta sección se describe el proceso de migración de las aplicaciones del conector de origen de Apache Kafka Connect que se ejecutan en las instalaciones o clústeres de Kafka Connect autogestionados que se ejecutan en AWS Amazon MSK Connect. La aplicación de conectores de origen de Kafka Connect almacena los desplazamientos en un tema con el nombre del valor establecido en la propiedad de configuración `offset.storage.topic`. A continuación, se muestran ejemplos de mensajes de desplazamientos de un conector JDBC que ejecuta dos tareas para importar datos de dos tablas distintas, denominadas `movies` y `shows`. La última fila importada de la tabla denominada “películas” tiene un ID primario de `18343`. La última fila importada de la tabla denominada “programas” tiene un ID primario de `732`. ``` ["jdbcsource",{"protocol":"1","table":"sample.movies"}] {"incrementing":18343} ["jdbcsource",{"protocol":"1","table":"sample.shows"}] {"incrementing":732} ``` Para migrar los conectores de origen a Amazon MSK Connect, siga los pasos a continuación: 1. Extrayendo las bibliotecas de conectores de sus clústeres en las instalaciones o autoadministrados de Kafta Connect, cree un [complemento personalizado](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html) de Amazon MSK Connect. 1. Cree las [propiedades del proceso](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-config-provider.html#msk-connect-config-providers-create-custom-config) de trabajo de Amazon MSK Connect y establezca las propiedades de `key.converter`, `value.converter` y `offset.storage.topic` en los mismos valores establecidos para el conector de Kafka que se ejecuta en su clúster existente de Kafka Connect. 1. Haga una solicitud de `PUT /connectors/connector-name/pause` en el clúster existente de Kafka Connect para pausar la aplicación de conectores en el clúster existente. 1. Asegúrese de que todas las tareas de la aplicación de conector estén completamente detenidas. Para detener las tareas, puede hacer una solicitud de `GET /connectors/connector-name/status` en el clúster existente de Kafka Connect o usar los mensajes del nombre de los temas establecido para la propiedad `status.storage.topic`. 1. Obtenga la configuración de los conectores del clúster existente. Para obtener la configuración de los conectores, puede hacer una solicitud de `GET /connectors/connector-name/config/` en el clúster existente o usar los mensajes del nombre de los temas establecido para la propiedad `config.storage.topic`. 1. Cree un nuevo [conector de Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html) con el mismo nombre que un clúster existente. Cree este conector usando el complemento personalizado de conectores que creó en el paso 1, las propiedades del proceso de trabajo que creó en el paso 2 y la configuración de los conectores que extrajo en el paso 5. 1. Cuando el estado del conector Amazon MSK sea `active`, consulte los registros para verificar que el conector haya empezado a importar datos del sistema de origen. 1. Haga una solicitud de `DELETE /connectors/connector-name` para eliminar el conector en el clúster existente. # Migración de los conectores de recepción a Amazon MSK Connect Los conectores de recepción son aplicaciones de Apache Kafka Connect que exportan datos desde Kafka a sistemas externos. En esta sección, se describe el proceso de migración de las aplicaciones del conector Sink de Apache Kafka Connect que se ejecutan en las instalaciones o clústeres de Kafka Connect autogestionados que se ejecutan en AWS Amazon MSK Connect. Los conectores de recepción de Kafka Connect utilizan la API de pertenencia a grupos de Kafka y almacenan desplazamientos en los mismos temas de `__consumer_offset` que las aplicaciones de consumidor habituales. Este comportamiento simplifica la migración del conector de recepción de un clúster autoadministrado a Amazon MSK Connect. Para migrar los conectores de recepción a Amazon MSK Connect, siga los pasos a continuación: 1. Extrayendo las bibliotecas de conectores de sus clústeres en las instalaciones o autoadministrados de Kafta Connect, cree un [complemento personalizado](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html) de Amazon MSK Connect. 1. Cree las [propiedades del proceso](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-config-provider.html#msk-connect-config-providers-create-custom-config) de trabajo de Amazon MSK Connect y establezca las propiedades de `key.converter` y `value.converter` en los mismos valores establecidos para el conector de Kafka que se ejecuta en su clúster existente de Kafka Connect. 1. Haga una solicitud de `PUT /connectors/connector-name/pause` en su clúster existente de Kafka Connect para pausar la aplicación de conectores en el clúster existente. 1. Asegúrese de que todas las tareas de la aplicación de conector estén completamente detenidas. Para detener las tareas, puede hacer una solicitud de `GET /connectors/connector-name/status` en el clúster existente de Kafka Connect o usar los mensajes del nombre de los temas establecido para la propiedad `status.storage.topic`. 1. Obtenga la configuración de conectores del clúster existente. Para obtener la configuración de los conectores, puede hacer una solicitud de `GET /connectors/connector-name/config` en el clúster existente o usar los mensajes del nombre de los temas establecido para la propiedad `config.storage.topic`. 1. Cree un nuevo [conector de Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html) con el mismo nombre que el clúster existente. Cree este conector usando el complemento personalizado de conectores que creó en el paso 1, las propiedades del proceso de trabajo que creó en el paso 2 y la configuración de los conectores que extrajo en el paso 5. 1. Cuando el estado del conector Amazon MSK sea `active`, consulte los registros para verificar que el conector haya empezado a importar datos del sistema de origen. 1. Haga una solicitud de `DELETE /connectors/connector-name` para eliminar el conector en el clúster existente. # Solución de problemas de Amazon MSK Connect La siguiente información le puede ayudar a solucionar los problemas que podrían presentarse con MSK Connect. También puede publicar el problema en [AWS re:Post](https://repost.aws/). **Connector no puede acceder a los recursos alojados de forma pública en Internet** Consulte [Habilitación del acceso a Internet para Amazon MSK Connect](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-internet-access.html). **El número de tareas en ejecución de Connector no es igual al número de tareas especificadas en tasks.max** Estas son algunas de las razones por las que un conector puede usar menos tareas que la configuración de tasks.max especificada: + Algunas implementaciones de conectores limitan la cantidad de tareas que se pueden utilizar. Por ejemplo, el conector Debezium para MySQL se limita a utilizar una sola tarea. + Cuando se utiliza el modo de capacidad con escalado automático, Amazon MSK Connect anula la propiedad tasks.max de un conector con un valor que es proporcional al número de trabajadores que trabajan en el conector y al número de trabajadores por trabajador. MCUs Si ha configurado el `maxAutoscalingTaskCount` parámetro opcional, el `tasks.max` valor no superará este límite. Para obtener más información, consulte [Comprender el número máximo de tareas de ajuste de escala automático](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-connectors.html#msk-connect-max-autoscaling-task-count). + En el caso de los conectores de recepción, el nivel de paralelismo (número de tareas) no puede ser superior al número de particiones temáticas. Si bien puede establecer el valor tasks.max en un tamaño mayor que ese valor, una sola partición nunca es procesada por más de una tarea a la vez. + En Kafka Connect 2.7.x, el asignador de particiones de consumo predeterminado es `RangeAssignor`. El comportamiento de este asignador consiste en entregar la primera partición de cada tema a un solo consumidor, la segunda partición de cada tema a un solo consumidor, etc. Esto significa que el número máximo de tareas activas de un conector de recepción con `RangeAssignor` es igual al número máximo de particiones consumidas en un solo tema. Si esto no funciona para su caso de uso, debería [crear una configuración de trabajo](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-create-custom-worker-config) en la que la propiedad `consumer.partition.assignment.strategy` esté establecida en un asignador de particiones de consumo más adecuado. Consulte [Interfaz de Kafka 2.7 ConsumerPartitionAssignor: *todas las clases de implementación conocidas*](https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.html).