[PIP-165] Auto release client useless connections

[poorbarcode]

Motivation

Currently, the Pulsar client keeps the connection even if no producers or consumers use this connection.
If a client produces messages to topic A and we have 3 brokers 1, 2, 3. Due to the bundle unloading(load manager)
topic ownership will change from A to B and finally to C. For now, the client-side will keep 3 connections to all 3 brokers.
We can optimize this part to reduce the broker side connections, the client should close the unused connections.

So a mechanism needs to be added to release unwanted connections.

Why are there idle connections?

1.When configuration maxConnectionsPerHosts is not set to 0, the connection is not closed at all.
The design is to hold a fixed number of connections per Host, avoiding frequent closing and creation.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java

Lines 325 to 335 in 7234911

	public void releaseConnection(ClientCnx cnx) {
	if (maxConnectionsPerHosts == 0) {
	//Disable pooling
	if (cnx.channel().isActive()) {
	if (log.isDebugEnabled()) {
	log.debug("close connection due to pooling disabled.");
	}
	cnx.close();
	}
	}
	}

2-1. When clients receive command-close, will reconnect immediately.
It's designed to make it possible to reconnect, rebalance, and unload.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java

Lines 122 to 141 in 7234911

	public void connectionClosed(ClientCnx cnx) {
	lastConnectionClosedTimestamp = System.currentTimeMillis();
	state.client.getCnxPool().releaseConnection(cnx);
	if (CLIENT_CNX_UPDATER.compareAndSet(this, cnx, null)) {
	if (!isValidStateForReconnection()) {
	log.info("[{}] [{}] Ignoring reconnection request (state: {})",
	state.topic, state.getHandlerName(), state.getState());
	return;
	}
	long delayMs = backoff.next();
	state.setState(State.Connecting);
	log.info("[{}] [{}] Closed connection {} -- Will try again in {} s",
	state.topic, state.getHandlerName(), cnx.channel(),
	delayMs / 1000.0);
	state.client.timer().newTimeout(timeout -> {
	log.info("[{}] [{}] Reconnecting after timeout", state.topic, state.getHandlerName());
	grabCnx();
	}, delayMs, TimeUnit.MILLISECONDS);
	}
	}

2-2. The broker will close client connections before writing ownership info to the ZK. Then clients will get deprecated broker address when it tries lookup.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 1282 to 1293 in 7234911

	producers.values().forEach(producer -> futures.add(producer.disconnect()));
	if (topicPublishRateLimiter != null) {
	try {
	topicPublishRateLimiter.close();
	} catch (Exception e) {
	log.warn("Error closing topicPublishRateLimiter for topic {}", topic, e);
	}
	}
	subscriptions.forEach((s, sub) -> futures.add(sub.disconnect()));
	if (this.resourceGroupPublishLimiter != null) {
	this.resourceGroupPublishLimiter.unregisterRateLimitFunction(this.getName());
	}

Goal

Automatically release connections that are no longer used.

• Scope

  • Pulsar client
    Contains connections used by consumers, Producers, and Transactions.

  • Pulsar proxy
    The connection between proxy and broker has two parts: For lookup commands; For consumers, producers commands and other commands.
    The connection "For consumers, producers commands and other commands" is managed by DirectProxyHandler, which holds the connection until the client is closed, so it does not affect of producers or consumers, These connections do not require additional closing.
    The connection "For lookup commands": When the proxy is configured metadataStoreUrl, the Lookup Command will select the registered broker by rotation training and create a connection. If we do not optimize the broker load balancing algorithm, all connections are considered useful connections.
    When the cluster is large, holds so many connections becomes redundant. Later, I will try to put forward other proposals to improve this phenomenon, so this proposal does not involve proxy connection release.

Approach

Periodically check for idle connections and close them.

Changes

API changes

ClientCnx added an idle check method to mark idle time.

/** Create time. **/
private final long createTime;
/** The time when marks the connection is idle. **/
private long IdleMarkTime;
/** The time when the last valid data was transmitted. **/
private long lastWorkTime;
/** Idle stat **/
private IdleState stat;
/** 
  * Check client connection is now free. This method may change the state to idle.
  * This method will not change the state to idle.
  */
public boolean doIdleCheck();
/** Get stat **/
public IdleState getIdleStat();
/** Change stat **/
public boolean setStat(IdleState originalIdleStat, IdleState newIdleStat);

public enum IdleState {
  /** The connection is in use. **/
  using,
  /** The connection is not in use. **/
  idle_marked,
  /** The connection will be released soon. **/
  before_release,
  /** The connection has already been released. **/
  released;
}

Configuration changes

We can set the check frequency and release rule for idle connections at ClientConfigurationData.

@ApiModelProperty(
        name = "connectionMaxIdleSeconds",
        value = "Release the connection if it is not used for more than [connectionMaxIdleSeconds] seconds. " 
                     + "If  [connectionMaxIdleSeconds] < 0, disabled the feature that auto release the idle connections"
)
private int connectionMaxIdleSeconds = 180;

@ApiModelProperty(
        name = "connectionIdleDetectionIntervalSeconds",
        value = "How often check idle connections"
)
private int connectionIdleDetectionIntervalSeconds = 60;

Implementation

• Pulsar client
  If no consumer, producer, or transaction uses the current connection, release it.
  • A fixed rate task in ConnectionPool, will do two things. 1: mark the connection idle 2: when reach max idle time release it
  • ConnectionCnx holds the Consumer ,Producer and TransactionClient, so we can rely on this information to determine if the connection is still in use

[poorbarcode]

@eolivelli @gaoran10 @gaozhangmin @315157973 @lhotari @michaeljmarshall I have sent an email, could you take a look.

Apache Mail Archives link

[poorbarcode]

@eolivelli @gaoran10 @gaozhangmin @315157973 @lhotari @michaeljmarshall I have start a vote and already send email, could you take a look.

Discuss Link: https://lists.apache.org/thread/t6h98qs2coc56z06tw38hdlljl67ft4n
Vote Link: https://lists.apache.org/thread/qfg9jfocpsc0tzy45ozgoothw1jpxk5g

[gaoran10]

Left a comment in discuss email list.

[gaoran10]

Maybe the config connectionMaxIdleSeconds could be as the switch, disable this feature if the connectionMaxIdleSeconds is a negative value.

[poorbarcode]

Maybe the config connectionMaxIdleSeconds could be as the switch, disable this feature if the connectionMaxIdleSeconds is a negative value.

Good idea. I changed it. Thanks @gaoran10

[liangyepianzhou]

Left a comment in discuss email list:)

[poorbarcode]

@eolivelli @315157973 Can you give a vote for this proposal, Thanks. https://lists.apache.org/thread/qfg9jfocpsc0tzy45ozgoothw1jpxk5g

[github-actions]

The issue had no activity for 30 days, mark with Stale label.