Mejores prácticas para los clientes de Apache Kafka

Disponibilidad del productor

• Establezca retries para indicar al productor que intente enviar nuevamente los mensajes con errores durante la conmutación por error del agente. Recomendamos que el valor utilizado sea un número entero máximo o un valor alto similar para la mayoría de los casos de uso. De lo contrario, se interrumpirá la alta disponibilidad de Kafka.

• Establezca delivery.timeout.ms para especificar el límite superior del tiempo total entre el envío de un mensaje y la recepción de una confirmación por parte del agente. Esto debe reflejar los requisitos empresariales respecto al tiempo de validez de un mensaje. Establezca un límite de tiempo lo suficientemente alto como para permitir que se realice la cantidad necesaria de reintentos para completar la conmutación por error. Recomendamos que se utilice un valor de 60 segundos o superior para la mayoría de los casos de uso.

• Establezca request.timeout.ms al tiempo máximo que una solicitud debe esperar antes de que se intente el reenvío. Recomendamos que se utilice un valor de 10 segundos o superior para la mayoría de los casos de uso.

• Establezca retry.backoff.ms para configurar el retraso entre los reintentos para evitar una avalancha de reintentos y un impacto en la disponibilidad. Recomendamos que se utilice un valor mínimo de 200 ms para la mayoría de los casos de uso.

• Establezca acks=all para configurar una alta durabilidad; esto debe alinearse con una configuración del servidor de RF=3 y min.isr=2 para garantizar que todas las particiones del ISR reconozcan la escritura. Cuando un solo agente está fuera de línea, este es el min.isr, es decir, 2.

Disponibilidad de los consumidores

• Establezca auto.offset.reset en latest inicialmente para los grupos de consumidores nuevos o creados nuevamente. Esto evita el riesgo de aumentar la carga de clústeres al consumir todo el tema.

• Establezca auto.commit.interval.ms cuando utilice enable.auto.commit. Recomendamos que se utilice un valor mínimo de 5 segundos para la mayoría de los casos de uso, para evitar el riesgo de una carga adicional.

• Implemente la gestión de excepciones en el código de procesamiento de mensajes del consumidor para gestionar los errores transitorios, por ejemplo, un disyuntor o una suspensión con un retroceso exponencial. Si no lo hace, puede producirse un error en la aplicación, lo que puede provocar un reequilibrio excesivo.

• Establezca isolation.level para controlar cómo se leen los mensajes transaccionales:

  Recomendamos que siempre se establezca read_uncommitted de forma implícita y predeterminada. Esto no aparece en algunas implementaciones de clientes.

  Recomendamos que utilice un valor de read_uncommitted al utilizar un almacenamiento en niveles.

• Establezca client.rack para usar la lectura de réplica más cercana. Recomendamos que se establezca en az id para minimizar los costos del tráfico de red y la latencia. Consulte Reduce network traffic costs of your Amazon MSK consumers with rack awareness.

Reequilibrios de los consumidores

• Establezca session.timeout.ms en un valor superior al tiempo de inicio de una aplicación, incluida cualquier fluctuación de inicio implementada. Recomendamos que se utilice un valor de 60 segundos para la mayoría de los casos de uso.

• Establezca heartbeat.interval.ms para afinar la forma en que el coordinador del grupo considera que un consumidor está en buen estado. Recomendamos que se utilice un valor de 10 segundos para la mayoría de los casos de uso.

• Establezca un enlace de cierre en su aplicación para cerrar sin problemas al consumidor en SIGTERM, en lugar de depender de los tiempos de espera de la sesión para identificar cuándo un consumidor abandona un grupo. Las aplicaciones de Kstream pueden establecer internal.leave.group.on.close en un valor de true.

• Establezca group.instance.id en un valor distinto dentro del grupo de consumidores. Idealmente, un nombre de host, task-id o pod-id. Recomendamos que siempre establezca esto para lograr comportamientos más deterministas y una mejor correlación de registros entre el cliente y el servidor durante la resolución de problemas.

• Establezca group.initial.rebalance.delay.ms en un valor que se alinee con el tiempo medio de implementación. Esto detiene los reequilibrios continuos durante la implementación.

• Establezca partition.assignment.strategy para utilizar asignadores fijos. Recomendamos que sea StickyAssignor o CooperativeStickyAssignor.

rendimiento del productor

• Establezca linger.ms para controlar el tiempo que un productor espera a que se llene un lote. Los lotes más pequeños son costosos desde el punto de vista computacional para Kafka, ya que generan más subprocesos y operaciones de E/S simultáneamente. Recomendamos que se utilicen los siguientes valores.

  Un valor mínimo de 5 ms para todos los casos de uso, incluida la baja latencia.

  Recomendamos que se utilice un valor más alto de 25 ms para la mayoría de los casos de uso.

  Recomendamos que nunca se utilice un valor de cero en los casos de uso con una latencia baja. (Por lo general, un valor de cero provoca latencia sin importar la sobrecarga de E/S).

• Establezca batch.size para controlar el tamaño del lote que se envía al clúster. Recomendamos que lo aumente a un valor de 64 KB o 128 KB.

• Establezca buffer.memory cuando utilice lotes de mayor tamaño. Recomendamos que se utilice un valor de 64 MB para la mayoría de los casos de uso.

• Establezca send.buffer.bytes para controlar el búfer TCP que se utiliza para recibir bytes. Recomendamos que se utilice un valor de -1 para que el sistema operativo pueda administrar este búfer cuando ejecute un productor en una red de alta latencia.

• Establezca compression.type para controlar la compresión de los lotes. Recomendamos que lz4 o zstd ejecuten un productor en una red de alta latencia.

rendimiento de los consumidores

• Establezca fetch.min.bytes para controlar el tamaño mínimo de recuperación que sea válido para reducir la cantidad de recuperaciones y la carga del clúster.

  Recomendamos que se utilice un valor mínimo de 32 bytes para todos los casos de uso.

  Recomendamos que se utilice un valor más alto de 128 bytes para la mayoría de los casos de uso.

• Establezca fetch.max.wait.ms para determinar cuánto tiempo esperará el consumidor antes de que se ignore fetch.min.bytes. Recomendamos que se utilice un valor de 1000 ms para la mayoría de los casos de uso.

• Recomendamos que el número de consumidores sea al menos igual al número de particiones para mejorar el paralelismo y la resiliencia. En algunas situaciones, puede optar por tener menos consumidores que el número de particiones por motivos de bajo rendimiento.

• Establezca receive.buffer.bytes para controlar el búfer TCP que se utiliza para recibir bytes. Recomendamos que se utilice un valor de -1 para que el sistema operativo pueda administrar este búfer cuando ejecute un consumidor en una red de alta latencia.

Conexiones de clientes

El ciclo de vida de las conexiones tiene un costo computacional y de memoria en un clúster Kafka. Si se crean demasiadas conexiones a la vez, se produce una carga que puede afectar a la disponibilidad de un clúster de Kafka. Este impacto en la disponibilidad suele provocar que las aplicaciones creen aún más conexiones, lo que provoca un error en cascada y una interrupción total. Se puede lograr un número elevado de conexiones si se crean a un ritmo razonable.

Recomendamos que se utilicen las siguientes medidas de mitigación para administrar las altas tasas de creación de conexiones:

• Asegúrese de que el mecanismo de implementación de sus aplicaciones no reinicie a todos los productores o consumidores a la vez, sino preferiblemente en lotes más pequeños.

• En el nivel de aplicación, el desarrollador debe garantizar que se produzca una fluctuación aleatoria (suspensión aleatoria) antes de crear un cliente de administración, un cliente productor o un cliente consumidor.

• En SIGTERM, al cerrar la conexión, se debe ejecutar una suspensión aleatoria para garantizar que los clientes de Kafka no se cierren todos al mismo tiempo. La suspensión aleatoria debe estar dentro del tiempo de espera antes de que se produzca SIGKILL.

  ejemplo Ejemplo A (Java)

  sleepInSeconds(randomNumberBetweenOneAndX);
                          this.kafkaProducer = new KafkaProducer<>(this.props);

  ejemplo Ejemplo B (Java)

  Runtime.getRuntime().addShutdownHook(new Thread(() -> {
      sleepInSeconds(randomNumberBetweenOneAndTwentyFive);
      kafkaProducer.close(Duration.ofSeconds(5));
  });

• En el nivel de aplicación, el desarrollador debe garantizar que los clientes se creen solo una vez por aplicación y con un patrón único. Por ejemplo, cuando se usa Lambda, el cliente debe crearse en un ámbito global y no en el controlador de métodos.

• Recomendamos que se controle la cantidad de conexiones para que se mantenga estable. creation/close/shiftLa conexión es normal durante las implementaciones y la conmutación por error del broker.

Métricas de productores

• byte-rate

• record-send-rate

• records-per-request-avg

• acks-latency-avg

• request-latency-avg

• request-latency-max

• record-error-rate

• record-retry-rate

• error-rate

Métricas de consumidores

• records-consumed-rate

• bytes-consumed-rate

• fetch-rate

• records-lag-max

• record-error-rate

• fetch-error-rate

• poll-rate

• rebalance-latency-avg

• commit-rate

Métricas comunes

• connection-close-rate

• connection-creation-rate

• connection-count