[improve][meta] Log a warning when ZK batch fails with connectionloss

[lhotari]

Motivation

A batch operation might cause a connection to break which is a severe issue. It's better to log a warning whenever this happens instead of silently ignoring it.

Modifications

• Log a warning when a batch operation fails with connectionloss.
• Add summary information to the log message.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[lhotari]

I might have encountered this issue recently when working on the /metrics endpoint issue.
A few times I got errors such as java.io.IOException: Packet len 15589885 is out of range! without any other proper explanation. More details of test setup in #22477 and #22494 comments. It seems that https://github.com/lhotari/pulsar-playground/blob/master/src/main/java/com/github/lhotari/pulsar/playground/TestScenarioCreateLongNamedTopics.java and https://github.com/lhotari/pulsar-playground/blob/master/src/main/java/com/github/lhotari/pulsar/playground/TestScenarioLoadAll.java would reproduce the problem I faced.

[lhotari]

The test

pulsar/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreBatchingTest.java

Lines 151 to 182 in 61296d9

	@Test(dataProvider = "impl")
	public void testBigBatchSize(String provider, Supplier<String> urlSupplier) throws Exception {
	@Cleanup
	MetadataStore store = MetadataStoreFactory.create(urlSupplier.get(), MetadataStoreConfig.builder()
	.batchingEnabled(true)
	.batchingMaxDelayMillis(1_000)
	.build());
	String key1 = newKey();
	// Create 20 MB of data and try to read it out
	int dataSize = 500 * 1024;
	byte[] payload = new byte[dataSize];
	int N = 40;
	List<CompletableFuture<Stat>> putFutures = new ArrayList<>();
	for (int i = 0; i < N; i++) {
	putFutures.add(store.put(key1 + "/" + i, payload, Optional.empty()));
	}
	FutureUtil.waitForAll(putFutures).join();
	List<CompletableFuture<Optional<GetResult>>> getFutures = new ArrayList<>();
	for (int i = 0; i < N; i++) {
	getFutures.add(store.get(key1 + "/" + i));
	}
	FutureUtil.waitForAll(getFutures).join();
	}

will reproduce the problem.
it will log "2024-04-23T17:03:51,416 - WARN - [main-EventThread:ZKMetadataStore@204] - Connection loss while executing batch operation of 40 GET entries of total data size of 910. Retrying individual operations one-by-one." with the changes in this PR.

[lhotari]

It seems that batching reads could cause more harm than benefit when the returned data exceeds jute.maxbuffer size.
This impacts stability a lot so there should be a way to disable batch reads completely.