kafka 参数篇

kafka Rebalancing 相关的参数有哪些

kafka 部门源码 （可忽略）

ConsumerConfig

1. public class ConsumerConfig extends AbstractConfig {
2. ...
3.     /** <code>max.poll.records</code> */
4.     public static final String MAX_POLL_RECORDS_CONFIG = "max.poll.records";
5.     private static final String MAX_POLL_RECORDS_DOC = "The maximum number of records returned in a single call to poll()."
6.         + " Note, that <code>" + MAX_POLL_RECORDS_CONFIG + "</code> does not impact the underlying fetching behavior."
7.         + " The consumer will cache the records from each fetch request and returns them incrementally from each poll.";
9.     /** <code>max.poll.interval.ms</code> */
10.     public static final String MAX_POLL_INTERVAL_MS_CONFIG = CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG;
11.     private static final String MAX_POLL_INTERVAL_MS_DOC = CommonClientConfigs.MAX_POLL_INTERVAL_MS_DOC;
12.    
13.     public static final String MAX_POLL_INTERVAL_MS_CONFIG = "max.poll.interval.ms";
14.     public static final String MAX_POLL_INTERVAL_MS_DOC = "The maximum delay between invocations of poll() when using "
15.                                                           + "consumer group management. This places an upper bound on the amount of time that the consumer can be idle "
16.                                                           + "before fetching more records. If poll() is not called before expiration of this timeout, then the consumer "
17.                                                           + "is considered failed and the group will rebalance in order to reassign the partitions to another member. "
18.                                                           + "For consumers using a non-null <code>group.instance.id</code> which reach this timeout, partitions will not be immediately reassigned. "
19.                                                           + "Instead, the consumer will stop sending heartbeats and partitions will be reassigned "
20.                                                           + "after expiration of <code>session.timeout.ms</code>. This mirrors the behavior of a static consumer which has shutdown.";
21.    
22.     /**
23.      * <code>session.timeout.ms</code>
24.      */
25.     public static final String SESSION_TIMEOUT_MS_CONFIG = CommonClientConfigs.SESSION_TIMEOUT_MS_CONFIG;
26.     private static final String SESSION_TIMEOUT_MS_DOC = CommonClientConfigs.SESSION_TIMEOUT_MS_DOC;
27.     public static final String SESSION_TIMEOUT_MS_CONFIG = "session.timeout.ms";
28.     public static final String SESSION_TIMEOUT_MS_DOC = "The timeout used to detect client failures when using "
29.                                                         + "Kafka's group management facility. The client sends periodic heartbeats to indicate its liveness "
30.                                                         + "to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, "
31.                                                         + "then the broker will remove this client from the group and initiate a rebalance. Note that the value "
32.                                                         + "must be in the allowable range as configured in the broker configuration by <code>group.min.session.timeout.ms</code> "
33.                                                         + "and <code>group.max.session.timeout.ms</code>.";
34.     /**
35.      * <code>heartbeat.interval.ms</code>
36.      */
37.     public static final String HEARTBEAT_INTERVAL_MS_CONFIG = CommonClientConfigs.HEARTBEAT_INTERVAL_MS_CONFIG;
38.     private static final String HEARTBEAT_INTERVAL_MS_DOC = CommonClientConfigs.HEARTBEAT_INTERVAL_MS_DOC;
40.     public static final String HEARTBEAT_INTERVAL_MS_CONFIG = "heartbeat.interval.ms";
41.     public static final String HEARTBEAT_INTERVAL_MS_DOC = "The expected time between heartbeats to the consumer "
42.                                                            + "coordinator when using Kafka's group management facilities. Heartbeats are used to ensure that the "
43.                                                            + "consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. "
44.                                                            + "The value must be set lower than <code>session.timeout.ms</code>, but typically should be set no higher "
45.                                                            + "than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.";
47. }

复制代码

Heartbeat

1. /**
2. * A helper class for managing the heartbeat to the coordinator
3. */
4. public final class Heartbeat {
5.     private final int maxPollIntervalMs;
6.     private final GroupRebalanceConfig rebalanceConfig;
7.     private final Time time;
8.     private final Timer heartbeatTimer;
9.     private final Timer sessionTimer;
10.     private final Timer pollTimer;
11.     private final Logger log;
12.     private final ExponentialBackoff retryBackoff;
14.     private volatile long lastHeartbeatSend = 0L;
15.     private volatile boolean heartbeatInFlight = false;
16.     private volatile long heartbeatAttempts = 0L;
18.     public Heartbeat(GroupRebalanceConfig config,
19.                      Time time) {
20.         if (config.heartbeatIntervalMs >= config.sessionTimeoutMs)
21.             throw new IllegalArgumentException("Heartbeat must be set lower than the session timeout");
22.         this.rebalanceConfig = config;
23.         this.time = time;
24.         this.heartbeatTimer = time.timer(config.heartbeatIntervalMs);
25.         this.sessionTimer = time.timer(config.sessionTimeoutMs);
26.         this.maxPollIntervalMs = config.rebalanceTimeoutMs;
27.         this.pollTimer = time.timer(maxPollIntervalMs);
28.         this.retryBackoff = new ExponentialBackoff(rebalanceConfig.retryBackoffMs,
29.                 CommonClientConfigs.RETRY_BACKOFF_EXP_BASE,
30.                 rebalanceConfig.retryBackoffMaxMs,
31.                 CommonClientConfigs.RETRY_BACKOFF_JITTER);
33.         final LogContext logContext = new LogContext("[Heartbeat groupID=" + config.groupId + "] ");
34.         this.log = logContext.logger(getClass());
35.     }
37.     private void update(long now) {
38.         heartbeatTimer.update(now);
39.         sessionTimer.update(now);
40.         pollTimer.update(now);
41.     }
43.     public void poll(long now) {
44.         update(now);
45.         pollTimer.reset(maxPollIntervalMs);
46.     }
48.     boolean hasInflight() {
49.         return heartbeatInFlight;
50.     }
52.     void sentHeartbeat(long now) {
53.         lastHeartbeatSend = now;
54.         heartbeatInFlight = true;
55.         update(now);
56.         heartbeatTimer.reset(rebalanceConfig.heartbeatIntervalMs);
58.         if (log.isTraceEnabled()) {
59.             log.trace("Sending heartbeat request with {}ms remaining on timer", heartbeatTimer.remainingMs());
60.         }
61.     }
63.     void failHeartbeat() {
64.         update(time.milliseconds());
65.         heartbeatInFlight = false;
66.         heartbeatTimer.reset(retryBackoff.backoff(heartbeatAttempts++));
68.         log.trace("Heartbeat failed, reset the timer to {}ms remaining", heartbeatTimer.remainingMs());
69.     }
71.     void receiveHeartbeat() {
72.         update(time.milliseconds());
73.         heartbeatInFlight = false;
74.         heartbeatAttempts = 0L;
75.         sessionTimer.reset(rebalanceConfig.sessionTimeoutMs);
76.     }
78.     boolean shouldHeartbeat(long now) {
79.         update(now);
80.         return heartbeatTimer.isExpired();
81.     }
82.    
83.     long lastHeartbeatSend() {
84.         return this.lastHeartbeatSend;
85.     }
87.     long timeToNextHeartbeat(long now) {
88.         update(now);
89.         return heartbeatTimer.remainingMs();
90.     }
92.     boolean sessionTimeoutExpired(long now) {
93.         update(now);
94.         return sessionTimer.isExpired();
95.     }
97.     void resetTimeouts() {
98.         update(time.milliseconds());
99.         sessionTimer.reset(rebalanceConfig.sessionTimeoutMs);
100.         pollTimer.reset(maxPollIntervalMs);
101.         heartbeatTimer.reset(rebalanceConfig.heartbeatIntervalMs);
102.     }
104.     void resetSessionTimeout() {
105.         update(time.milliseconds());
106.         sessionTimer.reset(rebalanceConfig.sessionTimeoutMs);
107.     }
109.     boolean pollTimeoutExpired(long now) {
110.         update(now);
111.         return pollTimer.isExpired();
112.     }
114.     long lastPollTime() {
115.         return pollTimer.currentTimeMs();
116.     }
117. }

复制代码

:1. session.timeout.ms

• 作用: 定义消费者在 Group Coordinator 中会话的超时时间。假如消费者在此时间内未发送心跳，Group Coordinator 会认为该消费者已失效，触发 Rebalancing。
  

   源码注释
  在利用Kafka消费者组管理机制时，用于检测客户端故障的会话超时时间。客户端会周期性地向Broker发送心跳信号以表明其存活状态。假如Broker在此会话超时到期前未收到心跳，则会将该客户端从消费者组中移除并触发再平衡。需要留意的是，该值必须处于Broker端设置参数group.min.session.timeout.ms（最小会话超时）和group.max.session.timeout.ms（最大会话超时）规定的允许范围内。
  

• 默认值: 通常为 10 秒（10000ms:）。
  
• 可能导致 Rebalancing 的缘故原由:
  
  
  
  • :消费者处置惩罚使命过重，导致无法按时发送心跳。
    
  • :网络延迟或故障，导致心跳无法及时到达 Group Coordinator。
    
  
  
• 优化建议: 根据消费者的处置惩罚本事和网络状态适当增大该值，但留意过大的值会延迟消费者失效的检测。
  

:2. heartbeat.interval.ms:

• 作用: 定义消费者发送心跳的时间间隔。
  

   源码注释：
  在利用Kafka消费者组管理机制时，与消费者协调器之间预期的心跳间隔时间。心跳机制用于维持消费者会话的活泼状态，并在新消费者到场或离开组时促进再平衡过程。该参数值必须小于session.timeout.ms（会话超时时间），通常建议设置为该值的1/3以下。可通过进一步调低此值来控制常规再平衡的预期相应时间。
  

• 默认值: 通常为 3 秒（3000ms:）。
  
• 可能导致 Rebalancing 的缘故原由:
  
  
  
  • :心跳间隔设置过长，导致消费者在: session.timeout.ms: :内未能发送足够的心跳。
    
  • :消费者处置惩罚使命过重或网络问题，导致心跳未能按时发送。
    
  
  
• 优化建议: 确保 heartbeat.interval.ms: 的值小于 session.timeout.ms: 的三分之一，并适当调解以平衡心跳频率和网络负载。
  

:3. max.poll.interval.ms:

• 作用: 定义消费者在两次调用 poll(): 方法之间的最大时间间隔。假如消费者在此时间内未调用 poll():，Group Coordinator 会认为该消费者已失效，触发 Rebalancing。 对于利用指定分组id (:group.instance.id