Cómo conectar Pub/Sub Lite a Apache Kafka

Nota: Pub/Sub Lite dejó de estar disponible. A partir del 18 de marzo de 2026, Pub/Sub Lite se desactivará.

• Clientes actuales: Pub/Sub Lite seguirá funcionando hasta el 18 de marzo de 2026. Si no usas Pub/Sub Lite en el período de 90 días anterior al 24 de septiembre de 2024 (del 26 de junio de 2024 al 24 de septiembre de 2024), no podrás acceder a Pub/Sub Lite a partir del 24 de septiembre de 2024.
• Clientes nuevos: Pub/Sub Lite ya no está disponible para clientes nuevos después del 24 de septiembre de 2024.

Puedes migrar tu servicio de Pub/Sub Lite al Servicio administrado de Google Cloud para Apache Kafka o a Pub/Sub.

En este documento, se describe cómo integrar Apache Kafka y Pub/Sub Lite con el conector de Kafka de grupos de Pub/Sub.

Nota: También puedes usar este conector con Pub/Sub. Consulta Cómo conectar Pub/Sub a Apache Kafka.

Acerca del conector de Kafka de grupos de Pub/Sub

Apache Kafka es una plataforma de código abierto para eventos de transmisión. Se suele usar en arquitecturas distribuidas para permitir la comunicación entre componentes con acoplamiento bajo. Pub/Sub Lite es un servicio administrado para enviar y recibir mensajes de forma asíncrona. Al igual que con Kafka, puedes usar Pub/Sub Lite para comunicarte entre componentes en tu arquitectura de nube.

El conector de Kafka de grupos de Pub/Sub te permite integrar estos dos sistemas. Los siguientes conectores se empaquetan en el archivo JAR del conector:

• El conector de receptor lee registros de uno o más temas de Kafka y los publica en Pub/Sub Lite.
• El conector de origen lee mensajes de un tema de Pub/Sub Lite y los publica en Kafka.

A continuación, se describen algunas situaciones en las que podrías usar el conector de Kafka de grupos de Pub/Sub:

• Estás migrando una arquitectura basada en Kafka a Google Cloud.
• Tienes un sistema de frontend que almacena eventos en Kafka fuera de Google Cloud, pero también usas Google Cloud para ejecutar algunos de tus servicios de backend, que deben recibir los eventos de Kafka.
• Recopila registros de una solución de Kafka local y envíalos a Google Cloud para el análisis de datos.
• Tienes un sistema de frontend que usa Google Cloud, pero también almacenas datos en las instalaciones con Kafka.

El conector requiere Kafka Connect, que es un framework para transmitir datos entre Kafka y otros sistemas. Para usar el conector, debes ejecutar Kafka Connect junto con tu clúster de Kafka.

En este documento, se da por sentado que conoces Kafka y Pub/Sub Lite. Para comenzar a usar Pub/Sub Lite, consulta Cómo publicar y recibir mensajes en Pub/Sub Lite con la consola de Google Cloud.

Comienza a usar el conector de Kafka de grupo de Pub/Sub

En esta sección, se explican las siguientes tareas:

1. Configura el conector de Kafka del grupo de Pub/Sub.
2. Envía eventos de Kafka a Pub/Sub Lite.
3. Envía mensajes de Pub/Sub Lite a Kafka.

Nota: En esta sección, se usa Kafka Connect en modo independiente, que es adecuado para pruebas. Para los sistemas de producción, se recomienda el modo distribuido. Para obtener información sobre cómo ejecutar Kafka Connect en modo distribuido, consulta Cómo ejecutar Kafka Connect en la documentación de Apache Kafka.

Requisitos previos

Instala Kafka

Sigue la guía de inicio rápido de Apache Kafka para instalar un nodo único de Kafka en tu máquina local. Completa estos pasos en la guía de inicio rápido:

1. Descarga la versión más reciente de Kafka y extráela.
2. Inicia el entorno de Kafka.
3. Crea un tema de Kafka.

Autenticar

El conector de Kafka de Pub/Sub Group debe autenticarse con Pub/Sub para enviar y recibir mensajes de Pub/Sub. Para configurar la autenticación, sigue estos pasos:

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
2. Install the Google Cloud CLI.

3. To initialize the gcloud CLI, run the following command:

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Create local authentication credentials for your user account:

   gcloud auth application-default login

6. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/pubsublite.admin

   gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE

   • Replace PROJECT_ID with your project ID.

   • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

   • Replace ROLE with each individual role.
7. Install the Google Cloud CLI.

8. To initialize the gcloud CLI, run the following command:

   gcloud init

9. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

10. Create local authentication credentials for your user account:

    gcloud auth application-default login

11. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/pubsublite.admin

    gcloud projects add-iam-policy-binding PROJECT_ID --member="USER_IDENTIFIER" --role=ROLE

    • Replace PROJECT_ID with your project ID.

    • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

    • Replace ROLE with each individual role.

Nota: Para las cargas de trabajo de producción, te recomendamos que uses la federación de identidades para cargas de trabajo para la autenticación.

Descarga el archivo JAR del conector

Descarga el archivo JAR del conector en tu máquina local. Para obtener más información, consulta Cómo adquirir el conector en la sección de lectura de GitHub.

Copia los archivos de configuración del conector

1. Clona o descarga el repositorio de GitHub del conector.

   git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git
   cd java-pubsub-group-kafka-connector
   

2. Copia el contenido del directorio config en el subdirectorio config de tu instalación de Kafka.

   cp config/* [path to Kafka installation]/config/
   

Estos archivos contienen la configuración del conector.

Actualiza la configuración de Kafka Connect

1. Navega al directorio que contiene el objeto binario de Kafka Connect que descargaste.
2. En el directorio binario de Kafka Connect, abre el archivo llamado config/connect-standalone.properties en un editor de texto.
3. Si el plugin.path property está comentado, quita el comentario.

4. Actualiza el plugin.path property para incluir la ruta de acceso al archivo JAR del conector.

   Ejemplo:

   plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar
   

5. Establece la propiedad offset.storage.file.filename en un nombre de archivo local. En el modo independiente, Kafka usa este archivo para almacenar datos de offset.

   Ejemplo:

   offset.storage.file.filename=/tmp/connect.offsets
   

Reenvía eventos de Kafka a Pub/Sub Lite

En esta sección, se describe cómo iniciar el conector de sink, publicar eventos en Kafka y, luego, leer los mensajes reenviados desde Pub/Sub Lite.

1. Usa Google Cloud CLI para crear una reserva de Pub/Sub Lite.

   gcloud pubsub lite-reservations create RESERVATION_NAME \
   --location=LOCATION \
   --throughput-capacity=4

   Reemplaza lo siguiente:

   • RESERVATION_NAME: Es el nombre de la reserva de Pub/Sub Lite.
   • LOCATION: Es la ubicación de la reserva.

2. Usa Google Cloud CLI para crear un tema de Pub/Sub Lite con una suscripción.

   gcloud pubsub lite-topics create LITE_TOPIC \
   --location=LOCATION \
   --partitions=2 \
   --per-partition-bytes=30GiB \
   --throughput-reservation=RESERVATION_NAME
   
   gcloud pubsub lite-subscriptions create LITE_SUBSCRIPTION \
   --location=LOCATION \
   --topic=LITE_TOPIC

   Reemplaza lo siguiente:

   • LITE_TOPIC: Es el nombre del tema de Pub/Sub Lite para recibir mensajes de Kafka.
   • LOCATION: Es la ubicación del tema. El valor debe coincidir con la ubicación de la reserva.
   • RESERVATION_NAME: Es el nombre de la reserva de Pub/Sub Lite.
   • LITE_SUBSCRIPTION: Es el nombre de una suscripción a Pub/Sub Lite para el tema.

3. Abre el archivo /config/pubsub-lite-sink-connector.properties en un editor de texto. Agrega valores para las siguientes propiedades, que están marcadas como "TODO" en los comentarios:

   topics=KAFKA_TOPICS
   pubsublite.project=PROJECT_ID
   pubsublite.location=LOCATION
   pubsublite.topic=LITE_TOPIC

   Reemplaza lo siguiente:

   • KAFKA_TOPICS: Es una lista separada por comas de los temas de Kafka desde los que se debe leer.
   • PROJECT_ID: Es el proyecto de Google Cloud que contiene tu tema de Pub/Sub Lite.
   • LOCATION: Es la ubicación del tema de Pub/Sub Lite.
   • LITE_TOPIC: Es el tema de Pub/Sub Lite para recibir mensajes de Kafka.

4. Desde el directorio de Kafka, ejecuta el siguiente comando:

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/pubsub-lite-sink-connector.properties
   

5. Sigue los pasos de la Guía de inicio rápido de Apache Kafka para escribir algunos eventos en tu tema de Kafka.

6. Suscríbete a la suscripción de Pub/Sub Lite con cualquiera de los métodos que se muestran en Cómo recibir mensajes de suscripciones Lite.

Reenvía mensajes de Pub/Sub Lite a Kafka

En esta sección, se describe cómo iniciar el conector de origen, publicar mensajes en Pub/Sub Lite y leer los mensajes reenviados desde Kafka.

1. Usa Google Cloud CLI para crear una reserva de Pub/Sub Lite.

   gcloud pubsub lite-reservations create RESERVATION_NAME \
   --location=LOCATION \
   --throughput-capacity=4

   Reemplaza lo siguiente:

   • RESERVATION_NAME: Es el nombre de la reserva de Pub/Sub Lite.
   • LOCATION: Es la ubicación de la reserva.

2. Usa Google Cloud CLI para crear un tema de Pub/Sub Lite con una suscripción.

   gcloud pubsub lite-topics create LITE_TOPIC \
   --location=LOCATION \
   --partitions=2 \
   --per-partition-bytes=30GiB \
   --throughput-reservation=RESERVATION_NAME
   
   gcloud pubsub lite-subscriptions create LITE_SUBSCRIPTION \
   --location=LOCATION \
   --topic=LITE_TOPIC

   Reemplaza lo siguiente:

   • LITE_TOPIC: Es el nombre del tema de Pub/Sub Lite.
   • LOCATION: Es la ubicación del tema. El valor debe coincidir con la ubicación de la reserva.
   • RESERVATION_NAME: Es el nombre de la reserva de Pub/Sub Lite.
   • LITE_SUBSCRIPTION: Es el nombre de una suscripción a Pub/Sub Lite para el tema.

3. Abre el archivo llamado /config/pubsub-lite-source-connector.properties en un editor de texto. Agrega valores para las siguientes propiedades, que están marcadas como "TODO" en los comentarios:

   topic=KAFKA_TOPIC
   pubsublite.project=PROJECT_ID
   pubsublite.location=LOCATION
   pubsublite.subscription=LITE_SUBSCRIPTION

   Reemplaza lo siguiente:

   • KAFKA_TOPIC: Son los temas de Kafka que recibirán los mensajes de Pub/Sub.
   • PROJECT_ID: Es el proyecto de Google Cloud que contiene tu tema de Pub/Sub.
   • LOCATION: Es la ubicación del tema de Pub/Sub Lite.
   • LITE_SUBSCRIPTION: Es el tema de Pub/Sub Lite.

4. Desde el directorio de Kafka, ejecuta el siguiente comando:

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/pubsub-lite-source-connector.properties
   

5. Publica mensajes en el tema de Pub/Sub Lite con cualquiera de los métodos que se muestran en Publica mensajes en temas de Lite.

6. Lee el mensaje de Kafka. Sigue los pasos de la Guía de inicio rápido de Apache Kafka para leer los mensajes del tema de Kafka.

Conversión de mensajes

Un registro de Kafka contiene una clave y un valor, que son arreglos de bytes de longitud variable. De manera opcional, un registro de Kafka también puede tener encabezados, que son pares clave-valor. Un mensaje de Pub/Sub Lite tiene los siguientes campos:

• key: Clave del mensaje (bytes)
• data: Datos del mensaje (bytes)
• attributes: Cero o más atributos. Cada atributo es un mapa (key,values[]). Un solo atributo puede tener varios valores.
• event_time: Es una marca de tiempo de evento opcional proporcionada por el usuario.

Kafka Connect usa convertidores para serializar claves y valores desde y hacia Kafka. Para controlar la serialización, establece las siguientes propiedades en los archivos de configuración del conector:

• key.converter: Es el convertidor que se usa para serializar las claves de registro.
• value.converter: Es el convertidor que se usa para serializar los valores de registro.

Conversión de Kafka a Pub/Sub Lite

El conector receptor convierte los registros de Kafka en mensajes de Pub/Sub Lite de la siguiente manera:

Registro de Kafka (SinkRecord)	Mensaje de Pub/Sub Lite
Clave	key
Valor	data
Encabezados	attributes
Marca de tiempo	eventTime
Tipo de marca de tiempo	attributes["x-goog-pubsublite-source-kafka-event-time-type"]
Tema	attributes["x-goog-pubsublite-source-kafka-topic"]
Partición	attributes["x-goog-pubsublite-source-kafka-offset"]
Desplazamiento	attributes["x-goog-pubsublite-source-kafka-partition"]

Las claves, los valores y los encabezados se codifican de la siguiente manera:

• Los esquemas nulos se tratan como esquemas de cadenas.
• Las cargas útiles de bytes se escriben directamente sin conversión.
• Las cargas útiles de cadenas, números enteros y números de punto flotante se codifican en una secuencia de bytes UTF-8.
• Todas las demás cargas útiles se codifican en un tipo de búfer de protocolo Value y, luego, se convierten en una cadena de bytes.
  • Los campos de cadena anidados se codifican en un Value de protobuf.
  • Los campos de bytes anidados se codifican en un Value de protobuf que contiene los bytes codificados en base64.
  • Los campos numéricos anidados se codifican como un número doble en un Value de protobuf.
  • No se admiten mapas con claves de array, mapa o struct.

Conversión de Pub/Sub Lite a Kafka

El conector de origen convierte los mensajes de Pub/Sub Lite en registros de Kafka de la siguiente manera:

Mensaje de Pub/Sub Lite	Registro de Kafka (SourceRecord)
key	Clave
data	Valor
attributes	Encabezados
event_time	Marca de tiempo Si event_time no está presente, se usa la hora de publicación.

Opciones de configuración

Además de las configuraciones que proporciona la API de Kafka Connect, el conector admite las siguientes configuraciones de Pub/Sub Lite.

Opciones de configuración del conector de sumidero

El conector de receptor admite las siguientes opciones de configuración.

Configuración	Tipo de datos	Descripción
connector.class	String	Obligatorio. Es la clase de Java del conector. Para el conector de sink de Pub/Sub Lite, el valor debe ser com.google.pubsublite.kafka.sink.PubSubLiteSinkConnector.
gcp.credentials.file.path	String	Opcional. Es la ruta de acceso a un archivo que almacena credenciales de Google Cloud para autenticar Pub/Sub Lite.
gcp.credentials.json	String	Opcional. Un blob JSON que contiene Google Cloud para autenticar Pub/Sub Lite.
pubsublite.location	String	Obligatorio. La ubicación del tema de Pub/Sub Lite
pubsublite.project	String	Obligatorio. La cuenta de Google Cloud que contiene el tema de Pub/Sub Lite
pubsublite.topic	String	Obligatorio. Es el tema de Pub/Sub Lite en el que se publicarán los registros de Kafka.
topics	String	Obligatorio. Una lista separada por comas de los temas de Kafka desde los que se debe leer.

Opciones de configuración del conector de origen

El conector de origen admite las siguientes opciones de configuración.

Configuración	Tipo de datos	Descripción
connector.class	String	Obligatorio. Es la clase de Java del conector. Para el conector de origen de Pub/Sub Lite, el valor debe ser com.google.pubsublite.kafka.source.PubSubLiteSourceConnector.
gcp.credentials.file.path	String	Opcional. Es la ruta de acceso a un archivo que almacena credenciales de Google Cloud para autenticar Pub/Sub Lite.
gcp.credentials.json	String	Opcional. Un blob JSON que contiene Google Cloud para autenticar Pub/Sub Lite.
kafka.topic	String	Obligatorio. El tema de Kafka que recibe mensajes de Pub/Sub Lite.
pubsublite.location	String	Obligatorio. La ubicación del tema de Pub/Sub Lite
pubsublite.partition_flow_control.bytes	Long	Es la cantidad máxima de bytes pendientes por partición de Pub/Sub Lite. Predeterminado: 20,000,000
pubsublite.partition_flow_control.messages	Long	Es la cantidad máxima de mensajes pendientes por partición de Pub/Sub Lite. Valor predeterminado: Long.MAX_VALUE
pubsublite.project	String	Obligatorio. El proyecto de Google Cloud que contiene el tema de Pub/Sub Lite.
pubsublite.subscription	String	Obligatorio. Es el nombre de la suscripción a Pub/Sub Lite desde la que se extraen los mensajes.

¿Qué sigue?

• Comprende las diferencias entre Kafka y Pub/Sub.
• Obtén más información sobre el conector de Kafka de grupos de Pub/Sub.
• Consulta el repositorio de GitHub del conector de Kafka de grupos de Pub/Sub.