[DISCUSS] [PIP-165] Auto release client useless connections

[Yubiao Feng]

I open a pip to discuss Auto release client useless connections, could you
help me review


## 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.

https://github.com/apache/pulsar/blob/72349117c4fd9825adaaf16d3588a695e8a9dd27/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java#L325-L335

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

https://github.com/apache/pulsar/blob/72349117c4fd9825adaaf16d3588a695e8a9dd27/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java#L122-L141

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.

https://github.com/apache/pulsar/blob/72349117c4fd9825adaaf16d3588a695e8a9dd27/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L1282-L1293

## Goal
Automatically release connections that are no longer used.

- Scope
  - **Pulsar client**
Contains connections used by consumers, Producers, and Transactions.

  - **Pulsar proxy**
Contains only the connection between Proxy and broker

## Approach
Periodically check for idle connections and close them.

## Changes

### API changes
**ClientCnx** added an idle check method to mark idle time.

```java
/** 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;
/** Stat. enumerated values: using, idle_marked, before_release, released**/
private int 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 boolen doIdleCheck();
/** Get stat **/
public int getStat();
/** Change stat **/
public int setStat(int originalStat, int newStat);
```

### Configuration changes
We can set the check frequency and release rule for idle connections at
`ClientConfigurationData`.

```java
@ApiModelProperty(
        name = "autoReleaseIdleConnectionsEnabled",
        value = "Do you want to automatically clean up unused connections"
)
private boolean autoReleaseIdleConnectionsEnabled = true;

@ApiModelProperty(
        name = "connectionMaxIdleSeconds",
        value = "Release the connection if it is not used for more than
[connectionMaxIdleSeconds] seconds"
)
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.

- **Pulsar proxy**
If the connection has not transmitted valid data for a long time, release
it.


Yubiao Feng
Thanks