protected GrpcClientChannel registerProducer(ProxyContext ctx, String topicName) {
        String clientId = ctx.getClientID();
        LanguageCode languageCode = LanguageCode.valueOf(ctx.getLanguage());

        GrpcClientChannel channel = this.grpcChannelManager.createChannel(ctx, clientId);
        // use topic name as producer group
        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(channel, clientId, languageCode, parseClientVersion(ctx.getClientVersion()));
        this.messagingProcessor.registerProducer(ctx, topicName, clientChannelInfo);
        TopicMessageType topicMessageType = this.messagingProcessor.getMetadataService().getTopicMessageType(ctx, topicName);
        if (TopicMessageType.TRANSACTION.equals(topicMessageType)) {
            this.messagingProcessor.addTransactionSubscription(ctx, topicName, topicName);
        }
        return channel;
    }
@Test
    public void testProducerHeartbeat() throws Throwable {
        ProxyContext context = createContext();

        this.sendProducerTelemetry(context);

        ArgumentCaptor<String> registerProducerGroupArgumentCaptor = ArgumentCaptor.forClass(String.class);
        ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class);
        doNothing().when(this.messagingProcessor).registerProducer(any(),
            registerProducerGroupArgumentCaptor.capture(),
            channelInfoArgumentCaptor.capture());

        ArgumentCaptor<String> txProducerGroupArgumentCaptor = ArgumentCaptor.forClass(String.class);
        ArgumentCaptor<String> txProducerTopicArgumentCaptor = ArgumentCaptor.forClass(String.class);
        doNothing().when(this.messagingProcessor).addTransactionSubscription(any(),
            txProducerGroupArgumentCaptor.capture(),
            txProducerTopicArgumentCaptor.capture()
        );

        when(this.metadataService.getTopicMessageType(any(), anyString())).thenReturn(TopicMessageType.TRANSACTION);

        HeartbeatResponse response = this.sendProducerHeartbeat(context);

        assertEquals(Code.OK, response.getStatus().getCode());

        assertEquals(Lists.newArrayList(TOPIC), registerProducerGroupArgumentCaptor.getAllValues());
        ClientChannelInfo clientChannelInfo = channelInfoArgumentCaptor.getValue();
        assertClientChannelInfo(clientChannelInfo, TOPIC);

        assertEquals(Lists.newArrayList(TOPIC), txProducerGroupArgumentCaptor.getAllValues());
        assertEquals(Lists.newArrayList(TOPIC), txProducerTopicArgumentCaptor.getAllValues());
    }