Lösung

Apache Kafka-Verbraucher gehören in der Regel einer Verbrauchergruppe an. Jeder Verbraucher in der Verbrauchergruppe erhält Nachrichten von einer anderen Teilmenge von Partitionen im Thema, wenn Folgendes passiert:

• Mehrere Verbraucher abonnieren ein Thema.
• Diese Verbraucher gehören derselben Verbrauchergruppe an.

Wenn ein Verbraucher den Cluster nicht erreichen kann, entfernt der Gruppenkoordinator den Verbraucher aus der Verbrauchergruppe. Dieser Prozess löst ein Neugewichtungs-Ereignis aus, das die folgenden Aktionen beinhaltet:

• Die verbleibenden Verbraucher werden von ihren Partitionen entlastet.
• Der Gruppenkoordinator verteilt die Partitionen des Themas an die verbleibenden Verbraucher neu.

Ihre Verbrauchergruppe kann unter den folgenden Bedingungen neu gewichtet werden:

• Sie haben den Partitionsbesitz von einem Verbraucher auf einen anderen in einer Benutzergruppe übertragen.
• Sie haben der Verbrauchergruppe einen neuen Verbraucher hinzugefügt.
• Ein Verbraucher wird abgeschaltet, stürzt ab oder verlässt die Verbrauchergruppe.
• Sie haben die Themen geändert und es erfolgt eine Neuausrichtung der Partitionen.
• Bei Abonnements für Verbrauchergruppen liegt ein Problem mit der Client-Konfiguration vor. Dies ist auf eine Diskrepanz zwischen den von der Gruppe abonnierten Themen und den Themen zurückzuführen, die den einzelnen Verbrauchern in der Gruppe zugewiesen wurden.

Die Verbraucher innerhalb derselben Verbrauchergruppe können den Datenverbrauch erst fortsetzen, wenn das Neugewichtungs-Ereignis abgeschlossen ist. Dies ist das standardmäßige Verhalten der Partitionszuweisung. Sie können dies vermeiden, indem Sie die Strategie für die Partitionszuweisung zu CooperativeStickyAssignor ändern.

Gehen Sie wie folgt vor, um zu verhindern, dass Ihre Verbrauchergruppe ständig neu gewichtet wird:

• Senken Sie entweder den Wert max.partition.fetch.bytes oder erhöhen Sie den Sitzungs-Timeout-Wert (session.timeout.ms) in der Verbraucherkonfiguration. Der Verbraucher muss poll() regelmäßig aufrufen, um ein Sitzungs-Timeout und eine anschließende Neugewichtung zu vermeiden. Wenn die Datenmenge, die ein einzelner poll() zurückgibt, groß ist, kann es lange dauern, bis der Verbraucher die Daten verarbeitet. Dies bedeutet, dass der Verbraucher nicht rechtzeitig zur nächsten Iteration der Abfrageschleife gelangt, um ein Sitzungs-Timeout zu vermeiden.
  Hinweis: Wenn Sie einen höheren Wert für session.timeout.ms festlegen, wird die Wahrscheinlichkeit einer versehentlichen Neugewichtung verringert. Es kann jedoch auch länger dauern, bis ein echter Fehler erkannt wird. Dieser Parameter steht im Zusammenhang mit heartbeat.interval.ms. Der Parameter heartbeat.interval.ms steuert die Frequenz, mit der die Methode KafkaConsumer poll() einen Herzschlag an den Gruppenkoordinator sendet. Der Parameter session.timeout.ms steuert jedoch, wie lange ein Verbraucher laufen darf, ohne einen Herzschlag zu senden.
  Nehmen wir zum Beispiel an, Sie verwenden Apache Kafka 0.10.1 oder höher und verarbeiten Datensätze, deren Verarbeitung länger dauert. In diesem Fall passen Sie max.poll.interval.ms an, um längere Verzögerungen zwischen der Abfrage neuer Datensätze zu bewältigen.
• Stellen Sie sicher, dass der Wert session.timeout.ms in der Verbraucherkonfiguration niedriger ist als der Wert group.max.session.timeout.ms in der Broker-Konfiguration.
• max.poll.interval.ms legt eine Obergrenze für die Zeit fest, für die der Verbraucher inaktiv sein darf, bevor weitere Datensätze abgerufen werden. In der Standardeinstellung ist dieser Wert auf 5 Minuten festgelegt. Wenn dieser Wert auf weniger als 5 Minuten eingestellt ist, erhöhen Sie ihn, um die Gefahr einer Neugewichtung zu verringern. Sie können zudem in Betracht ziehen, max.poll.records zusammen mit max.poll.interval.ms zu verringern.
• heartbeat.interval.ms ist die erwartete Zeit zwischen den Herzschlägen zum Verbraucherkoordinator, wenn Sie die Gruppenverwaltungsfunktionen von Kafka nutzen. Herzschläge werden verwendet, um sicherzustellen, dass die Sitzung des Verbrauchers aktiv bleibt. Sie erleichtern die Neugewichtung, wenn neue Verbraucher der Gruppe beitreten oder sie verlassen. Dieser Wert muss auf einen Wert gesetzt werden, der niedriger als session.timeout.ms ist. Normalerweise muss dieser Wert auf einen Wert gesetzt werden, der nicht höher als ein Drittel von session.timeout.ms ist. Sie können den Wert heartbeat.interval.ms deutlich senken, um die erwartete Zeit für normale Neugewichtungen zu kontrollieren.
• Wenn Sie kürzlich eine Neuzuweisung von Partitionen vorgenommen haben, die Änderungen an Partitionen in einem der abonnierten Themen der Benutzergruppe beinhaltet, kann es sein, dass die Benutzergruppe neu gewichtet wird. Das liegt daran, dass die beteiligten Partitionen verschoben oder verändert werden. In diesem Fall sollten Sie den Gruppenkoordinator oder andere Kafka-Broker nicht neu starten. Sie müssen warten, bis die Neuzuweisung der Partition abgeschlossen ist, bevor Sie versuchen, die Benutzergruppe an der Neugewichtung zu hindern. Es hat sich bewährt, Partitionsneuzuweisungen in Zeiten mit geringem Datenverkehr vorzunehmen.

In einigen Fällen sehen Sie möglicherweise die folgenden Informationen in den Amazon MSK-Brokerprotokollen:

[2023-03-01 01:23:45,678] INFO [GroupCoordinator 1]: Preparing to rebalance group amazon.msk.canary.group.broker-1 in state PreparingRebalance with old generation 382660 (__consumer_offsets-21) (reason: Adding new member consumer-amazon.msk.canary.group.broker-1-xxxx-xxxx-xxxx-xxxx-xxxx-xxxx with group instance id None) (kafka.coordinator.group .GroupCoordinator)

Diese Meldung weist darauf hin, dass sich amazon.msk.canary.group.broker-N im Status PreparingRebalance befindet.

amazon.msk.canary.group.broker-N-Gruppen sind interne Benutzergruppen, die regelmäßig hinzugefügt oder entfernt werden, um den Clusterstatus und die Diagnosemetriken zu überprüfen. Diese Gruppen sind vernachlässigbar groß und können nicht gelöscht werden. Sie können diese Meldung ignorieren.

Verwandte Informationen

Verbrauchergruppe bleibt im Status PreparingRebalance hängen