@Override
    public String getTopic() {
        StringBuilder sb = new StringBuilder();
        if (StringUtils.isNotBlank(config.getMm2SourceAlias())) {
            sb.append(config.getMm2SourceAlias()).append(config.getMm2SourceSeparator());
        }

        if (StringUtil.isNotBlank(config.getNamespace())) {
            sb.append(config.getNamespace()).append("-");
        }
        sb.append(getPlainTopic());
        return sb.toString();
    }
@Test
    public void testGetTopic() {
        KafkaFetcherConfig config = new KafkaFetcherConfig();

        MockModuleManager manager = new MockModuleManager() {
            @Override
            protected void init() {
            }
        };
        String plainTopic = config.getTopicNameOfTracingSegments();

        MockKafkaHandler kafkaHandler = new MockKafkaHandler(plainTopic, manager, config);

        //  unset namespace and mm2
        assertEquals(kafkaHandler.getTopic(), plainTopic);

        //set namespace only
        String namespace = "product";
        config.setNamespace(namespace);
        assertEquals(namespace + "-" + plainTopic, kafkaHandler.getTopic());

        //set mm2 only
        config.setNamespace("");
        String mm2SourceAlias = "DC1";
        config.setMm2SourceAlias(mm2SourceAlias);
        String mm2SourceSeparator = ".";
        config.setMm2SourceSeparator(mm2SourceSeparator);
        assertEquals(mm2SourceAlias + mm2SourceSeparator + plainTopic, kafkaHandler.getTopic());

        //set namespace and mm2
        config.setNamespace(namespace);
        config.setMm2SourceAlias(mm2SourceAlias);
        config.setMm2SourceSeparator(mm2SourceSeparator);
        assertEquals(mm2SourceAlias + mm2SourceSeparator + namespace + "-" + plainTopic, kafkaHandler.getTopic());

    }