@Override
    public void invoke(IN value, Context context) throws Exception {
        bufferLock.lock();
        try {
            // TODO this implementation is not very effective,
            //  optimize this with MemorySegment if needed
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(baos);
            serializer.serialize(value, wrapper);
            invokingRecordBytes = baos.size();

            if (invokingRecordBytes > maxBytesPerBatch) {
                throw new RuntimeException(
                        "Record size is too large for CollectSinkFunction. Record size is "
                                + invokingRecordBytes
                                + " bytes, "
                                + "but max bytes per batch is only "
                                + maxBytesPerBatch
                                + " bytes. "
                                + "Please consider increasing max bytes per batch value by setting "
                                + CollectSinkOperatorFactory.MAX_BATCH_SIZE.key());
            }

            if (currentBufferBytes + invokingRecordBytes > bufferSizeLimitBytes) {
                bufferCanAddNextResultCondition.await();
            }
            buffer.add(baos.toByteArray());
            currentBufferBytes += baos.size();
        } finally {
            bufferLock.unlock();
        }
    }
@Test
    void testInvalidToken() throws Exception {
        functionWrapper.openFunction();
        for (int i = 0; i < 6; i++) {
            functionWrapper.invoke(i);
        }
        String version = initializeVersion();
        functionWrapper.sendRequestAndGetResponse(version, 4);

        // invalid token
        CollectCoordinationResponse response =
                functionWrapper.sendRequestAndGetResponse(version, 3);
        assertResponseEquals(response, version, 0, Collections.emptyList());

        functionWrapper.closeFunctionNormally();
    }