[cleanup][test] Remove PortManager and use kernel-assigned ports everywhere

[merlimat]

Motivation

PortManager (in pulsar-common) and the inner BasePortManager of LocalBookkeeperEnsemble allocated "free" ports via ServerSocket(0) and tracked them in a JVM-level lock set. The lock was JVM-only, so the OS-level port could still be grabbed by another process between allocation and bind, producing flaky BindExceptions in tests (most recently ShadowTopicRealBkTest.setup).

The class has outlived its usefulness — modern BookKeeper identifies bookies by bookieId, not host:port, so kernel-assigned ports (port 0) work across the board. The "lock" never actually prevented OS-level collisions anyway.

Modifications

Removed

• pulsar-common/.../PortManager.java and its test.
• pulsar-package-management/.../test/PortManager.java (unused duplicate).
• BasePortManager inner class in LocalBookkeeperEnsemble.
• All bkBasePort constructor variants of LocalBookkeeperEnsemble.
• The Supplier<Integer> portManager parameter from LocalBookkeeperEnsemble — every caller was passing () -> 0. Bookies now always bind to kernel-assigned ports internally.
• --bookkeeper-port CLI option from pulsar standalone (and its setter, getter, builder method, plus the test calls). Multi-bookie standalone gets kernel-assigned ports.
• useDynamicBrokerPorts() toggle in MultiBrokerBaseTest (now always dynamic).
• bkPort field in BKCluster.BKClusterConf.

Refactored

• BKCluster: bookies bind to port 0; bookieId is derived from the data dir hash so multiple cluster instances on a shared metadata store don't collide on cookie validation, while restarts on the same dir keep cookies stable. The unused cookie-parsing port-recovery logic is dropped.
• MultiBrokerBaseTest: mainBrokerPort and additionalBrokerPorts are populated post-start from pulsar.getBrokerListenPort() instead of pre-allocated.
• PulsarTestContext.handlePreallocatePorts: inlined ServerSocket(0) allocation. Still needed by tests that build advertised-listener URLs before the broker starts (those URLs require a known port up front).
• 17 test files updated: PortManager.nextLockedFreePort() calls replaced with 0 (let kernel pick) or inline ServerSocket(0) (only where pre-allocation is genuinely required, e.g. for advertised-listener URLs).
• PulsarStandaloneTest.testStandaloneWithRocksDB: assertion switched from getBookiePort() (now returns kernel-assigned ports that change per restart) to getBookieId() (the persistent BK identity).

Behavior change

--bookkeeper-port is gone from pulsar standalone. Bookies get kernel-assigned ports — no impact for normal usage since clients only connect to the broker, not directly to bookies.

Verifying this change

Locally exercised the affected tests:

• LocalBookkeeperEnsembleTest (2/2)
• ShadowTopicRealBkTest (1/1)
• PulsarStandaloneTest (5/5)
• EndToEndTest for BKCluster (10 passed, 1 skipped)
• AdvertisedListenersTest, LoadManagerFailFastTest, AdvertisedListenersMultiBrokerLeaderElectionTest

Full compileJava and compileTestJava clean.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API: removes --bookkeeper-port CLI option from pulsar standalone
• The schema: no
• The default values of configurations: no
• The threading model: no
• The binary protocol: no
• The REST endpoints: no
• The admin CLI options: --bookkeeper-port removed from pulsar standalone
• The metrics: no
• Anything that affects deployment: no

[lhotari]

There's still a need for PortManager in cases where ephemeral ports (0) can't be used. Where they can, ephemeral is obviously preferred.

To avoid collisions for assigned ports across multiple processes, PortManager should allocate from outside the ephemeral range. On Linux the default emphemeral range is 32768–60999:

❯ sysctl net.ipv4.ip_local_port_range
net.ipv4.ip_local_port_range = 32768    60999

Proposal: reserve a block per JVM starting at a fixed offset. Each instance claims block n by binding a server socket on 20000 + (n * 1000), giving it 999 usable ports in that range. That handles coordination across multiple test JVMs without a shared registry. If ports run out, a new block could be allocated.

On release, PortManager should verify the port is actually free before returning it to the pool — sockets can linger in TIME_WAIT after close, so the free-list update needs to handle async release rather than marking the port available immediately.

[merlimat]

@lhotari thanks — restored PortManager and addressed both improvements you suggested.

The new implementation in b0e844dbab0:

• Allocates from a 1000-port block outside the ephemeral range. Each JVM claims its block by binding a "lock" ServerSocket on the block's base port for the JVM's lifetime; other JVMs that hit the same range observe the bind failure and walk to the next block. Default range is [20000, 33000) which gives 13 blocks before exhaustion — should be plenty for parallel test JVMs.
• releaseLockedPort now moves the port to a pending-release set rather than returning it to the pool immediately. The next allocation rebinds and only reclaims pending-release ports that bind successfully, so TIME_WAIT sockets don't get handed out.

Wired it back into the call sites that genuinely need pre-allocation: ServiceUrlQuarantineTest, ExtensibleLoadManagerImplWithAdvertisedListenersTest, ModularLoadManagerImplTest.testOwnBrokerZnodeByMultipleBroker, AdvertisedListenersTest, SimpleProtocolHandlerTestsBase, SimpleProxyExtensionTestBase, and PulsarTestContext.handlePreallocatePorts (the preallocatePorts(true) opt-in is preserved). Everything else still uses port 0 + read-back via getBrokerListenPort() / getZookeeperPort().

[lhotari]

LGTM