Custom payload with null keys is failing in an odd spot

[absurdfarce]

Test app using Java driver 4.19.1 against Cassandra 5.0.6. Should be protocol v5.

public class KeyspaceCount {

    public static void main(String[] args) {

        try (CqlSession session = CqlSession.builder().build()) {

            SimpleStatement stmt = SimpleStatement.builder("select release_version from system.local").build();
            HashMap<String, ByteBuffer> map = new HashMap<>();
            map.put(null, ByteBuffer.allocate(0));

            ResultSet rs = session.execute(stmt.setCustomPayload(map));
            Row row = rs.one();

            assert row != null;
            String releaseVersion = row.getString("release_version");
            System.out.printf("Cassandra version is: %s%n", releaseVersion);
        }
    }
}

Exception in thread "main" com.datastax.oss.driver.api.core.AllNodesFailedException: All 1 node(s) tried for the query failed (showing first 1 nodes, use getAllErrors() for more): Node(endPoint=/127.0.0.1:9042, hostId=9ae46b97-1648-4c4b-a
635-de8d83c2b204, hashCode=502d0989): [java.lang.NullPointerException: Cannot invoke "String.length()" because "st" is null]                                                                                                                  
        at com.datastax.oss.driver.api.core.AllNodesFailedException.copy(AllNodesFailedException.java:143)                                                                                                                                    
        at com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures.getUninterruptibly(CompletableFutures.java:154)                                                                                                           
        at com.datastax.oss.driver.internal.core.cql.CqlRequestSyncProcessor.process(CqlRequestSyncProcessor.java:55)                                                                                                                         
        at com.datastax.oss.driver.internal.core.cql.CqlRequestSyncProcessor.process(CqlRequestSyncProcessor.java:32)                                                                                                                         
        at com.datastax.oss.driver.internal.core.session.DefaultSession.execute(DefaultSession.java:234)
        at com.datastax.oss.driver.api.core.cql.SyncCqlSession.execute(SyncCqlSession.java:56)
        at com.ibm.KeyspaceCount.main(KeyspaceCount.java:39)
        Suppressed: java.lang.NullPointerException: Cannot invoke "String.length()" because "st" is null
                at com.datastax.oss.protocol.internal.PrimitiveSizes.encodedUTF8Length(PrimitiveSizes.java:96)
                at com.datastax.oss.protocol.internal.PrimitiveSizes.sizeOfString(PrimitiveSizes.java:35)
                at com.datastax.oss.protocol.internal.PrimitiveSizes.sizeOfBytesMap(PrimitiveSizes.java:87)
                at com.datastax.oss.protocol.internal.FrameCodec.encodedBodySize(FrameCodec.java:265)
                at com.datastax.oss.protocol.internal.SegmentBuilder.addFrame(SegmentBuilder.java:75)
                at com.datastax.oss.driver.internal.core.protocol.FrameToSegmentEncoder.write(FrameToSegmentEncoder.java:58)
                at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:893)
                at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:875)
                at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:984)
                at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:868)
                at io.netty.handler.timeout.IdleStateHandler.write(IdleStateHandler.java:307)
                at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:891)
                at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:875)
                at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:984)
                at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:868)
                at com.datastax.oss.driver.internal.core.channel.InFlightHandler.write(InFlightHandler.java:153)
                at com.datastax.oss.driver.internal.core.channel.InFlightHandler.write(InFlightHandler.java:110)
                at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:891)
                at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:875)
                at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:984)
                at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:868)
                at io.netty.channel.DefaultChannelPipeline.write(DefaultChannelPipeline.java:964)
                at io.netty.channel.AbstractChannel.write(AbstractChannel.java:300)
                at com.datastax.oss.driver.internal.core.channel.DefaultWriteCoalescer$Flusher.runOnEventLoop(DefaultWriteCoalescer.java:102)
                at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
                at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
                at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
                at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
                at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
                at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
                at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
                at java.base/java.lang.Thread.run(Thread.java:833)

I expected this to fail: the v4 spec (when custom payload was added) defined custom payload a bytes map which has strings for keys and bytes for values. There's no obvious way to implement null strings (while we can implement null bytes) so this kind of thing should fail... but it seems like it should fail earlier and/or provide better guidance to the user than the above.

[absurdfarce]

Note to self: raw write in the driver uses Channel.write() where Channel is a Netty construct. The integration back into native-protocol should come from the Netty pipeline.