@merlimat PTAL

@lhotari please check the answer from @merlimat
#10210 (comment)

Probably this is the minimal fix we can do in ClientCtx, that is to eagerly initialize all of the fields:

@Override
    protected void handleGetSchemaResponse(CommandGetSchemaResponse commandGetSchemaResponse) {
        checkArgument(state == State.Ready);

        long requestId = commandGetSchemaResponse.getRequestId();
        commandGetSchemaResponse.getSchema().getSchemaData();
        commandGetSchemaResponse.getSchema().getName();
        commandGetSchemaResponse.getSchema().getPropertiesList()
        commandGetSchemaResponse.getSchema().getType();
        CompletableFuture<CommandGetSchemaResponse> future = (CompletableFuture<CommandGetSchemaResponse>) pendingRequests
                .remove(requestId);
        if (future == null) {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), requestId);
            return;
        }
        future.complete(commandGetSchemaResponse);
    }

@lhotari please check the answer from @merlimat
#10210 (comment)

yes I checked it.

Probably this is the minimal fix we can do in ClientCtx, that is to eagerly initialize all of the fields:

@Override
    protected void handleGetSchemaResponse(CommandGetSchemaResponse commandGetSchemaResponse) {
        checkArgument(state == State.Ready);

        long requestId = commandGetSchemaResponse.getRequestId();
        commandGetSchemaResponse.getSchema().getSchemaData();
        commandGetSchemaResponse.getSchema().getName();
        commandGetSchemaResponse.getSchema().getPropertiesList()
        commandGetSchemaResponse.getSchema().getType();
        CompletableFuture<CommandGetSchemaResponse> future = (CompletableFuture<CommandGetSchemaResponse>) pendingRequests
                .remove(requestId);
        if (future == null) {
            log.warn("{} Received unknown request id from server: {}", ctx.channel(), requestId);
            return;
        }
        future.complete(commandGetSchemaResponse);
    }

This won't work since there is no eager initialization. Please take a look at the source code of LightProto generated classes.

A way to do eager initialization would be to move the code for creating SchemaInfo from sendGetSchema method to handleGetSchemaResponse. I experimented with that here: lhotari@431b41c .
It works, but the problem is that Pulsar Proxy depends on the current way and that would break. That's the reason why I came back to the solution that I have in this PR.

For instance on Schema struct when you call "getName" the String is created and the ByteBuf is decoded:

public String getName() {
		if (!hasName()) {
			throw new IllegalStateException("Field 'name' is not set");
		}
		if (name == null) {
			name = LightProtoCodec.readString(_parsedBuffer, _nameBufferIdx, _nameBufferLen);
		}
		return name;
	}

with my approach you eagerly create the "String" for name and then the problem is fixed

your fix works but it needs to refactor a bit the structures that we are passing from method to method.

if LightProto had an utility function to eagerly decode the ByteBufs we could use that feature, without the need of calling every method that needs an explicit call for initialization (ByteBuf decode I mean)

@merlimat ?

with my approach you eagerly create the "String" for name and then the problem is fixed

For Strings it's fine, but the approach doesn't seem to work for getSchemaData(). Did you check the source code of that method?

The protobuf object is only valid within the context of the IO thread that is deserializing it. If we need some of the fields to be used from different threads, these values have to be copied out. eg: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L866

/pulsarbot run-failure-checks

I have another idea (that I am testing locally)

We can simply "clone" the CommandGetSchemaResponse inside handleGetSchemaResponse, this way we decode the ByteBuf and we return a safe object to the caller

 @Override
    protected void handleGetSchemaResponse(CommandGetSchemaResponse commandGetSchemaResponse) {
        log.info("handleGetSchemaResponse {}", commandGetSchemaResponse);
        checkArgument(state == State.Ready);
        CommandGetSchemaResponse clone = new CommandGetSchemaResponse();
        clone.copyFrom(commandGetSchemaResponse);
        commandGetSchemaResponse = clone;

We can simply "clone" the CommandGetSchemaResponse inside handleGetSchemaResponse, this way we decode the ByteBuf and we return a safe object to the caller

@eolivelli Good idea. I revisited the changes and also found 2 other locations where a copy is needed.

Can we add test cases?

@sijie Do you have a proposal of how to test this?

@merlimat I have revisited the change. PTAL

@sijie @merlimat I have this integration test that fails due to this problem, I believe there is no strong need to add a test here
https://github.com/apache/pulsar/pull/10211/checks?check_run_id=2362726281