@Override
    protected RemotingCommand processRequest0(ChannelHandlerContext ctx, RemotingCommand request,
        ProxyContext context) throws Exception {
        PullMessageRequestHeader requestHeader = (PullMessageRequestHeader) request.decodeCommandCustomHeader(PullMessageRequestHeader.class);
        int sysFlag = requestHeader.getSysFlag();
        if (!PullSysFlag.hasSubscriptionFlag(sysFlag)) {
            ConsumerGroupInfo consumerInfo = messagingProcessor.getConsumerGroupInfo(context, requestHeader.getConsumerGroup());
            if (consumerInfo == null) {
                return RemotingCommand.buildErrorResponse(ResponseCode.SUBSCRIPTION_NOT_LATEST,
                    "the consumer's subscription not latest");
            }
            SubscriptionData subscriptionData = consumerInfo.findSubscriptionData(requestHeader.getTopic());
            if (subscriptionData == null) {
                return RemotingCommand.buildErrorResponse(ResponseCode.SUBSCRIPTION_NOT_EXIST,
                    "the consumer's subscription not exist");
            }
            requestHeader.setSysFlag(PullSysFlag.buildSysFlagWithSubscription(sysFlag));
            requestHeader.setSubscription(subscriptionData.getSubString());
            requestHeader.setExpressionType(subscriptionData.getExpressionType());
            request.writeCustomHeader(requestHeader);
            request.makeCustomHeaderToNet();
        }
        long timeoutMillis = requestHeader.getSuspendTimeoutMillis() + Duration.ofSeconds(10).toMillis();
        return request(ctx, request, context, timeoutMillis);
    }
@Test
    public void testPullMessageWithoutSub() throws Exception {
        when(messagingProcessorMock.getConsumerGroupInfo(any(), eq(group)))
            .thenReturn(consumerGroupInfoMock);
        SubscriptionData subscriptionData = new SubscriptionData();
        subscriptionData.setSubString(subString);
        subscriptionData.setExpressionType(type);
        when(consumerGroupInfoMock.findSubscriptionData(eq(topic)))
            .thenReturn(subscriptionData);

        PullMessageRequestHeader header = new PullMessageRequestHeader();
        header.setTopic(topic);
        header.setConsumerGroup(group);
        header.setQueueId(0);
        header.setQueueOffset(0L);
        header.setMaxMsgNums(16);
        header.setSysFlag(PullSysFlag.buildSysFlag(true, false, false, false));
        header.setCommitOffset(0L);
        header.setSuspendTimeoutMillis(1000L);
        header.setSubVersion(0L);
        header.setBrokerName(brokerName);

        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, header);
        request.makeCustomHeaderToNet();
        RemotingCommand expectResponse = RemotingCommand.createResponseCommand(ResponseCode.NO_MESSAGE, "success");
        PullMessageRequestHeader newHeader = new PullMessageRequestHeader();
        newHeader.setTopic(topic);
        newHeader.setConsumerGroup(group);
        newHeader.setQueueId(0);
        newHeader.setQueueOffset(0L);
        newHeader.setMaxMsgNums(16);
        newHeader.setSysFlag(PullSysFlag.buildSysFlag(true, false, true, false));
        newHeader.setCommitOffset(0L);
        newHeader.setSuspendTimeoutMillis(1000L);
        newHeader.setSubVersion(0L);
        newHeader.setBrokerName(brokerName);
        newHeader.setSubscription(subString);
        newHeader.setExpressionType(type);
        RemotingCommand matchRequest = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, newHeader);
        matchRequest.setOpaque(request.getOpaque());
        matchRequest.makeCustomHeaderToNet();

        ArgumentCaptor<RemotingCommand> captor = ArgumentCaptor.forClass(RemotingCommand.class);
        when(messagingProcessorMock.request(any(), eq(brokerName), captor.capture(), anyLong()))
            .thenReturn(CompletableFuture.completedFuture(expectResponse));
        RemotingCommand response = pullMessageActivity.processRequest0(ctx, request, null);
        assertThat(captor.getValue().getExtFields()).isEqualTo(matchRequest.getExtFields());
        assertThat(response).isNull();
        verify(ctx, times(1)).writeAndFlush(eq(expectResponse));
    }