public Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue> getMetadataFields() {
    return _metadataFields;
  }
@Test
  public void testParsingMetadataConfigWithoutConfiguredFields() {
    Map<String, String> streamConfigMap = getCommonStreamConfigMap();
    streamConfigMap.put(
        StreamConfigProperties.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.METADATA_POPULATE), "true");
    StreamConfig streamConfig = new StreamConfig(TABLE_NAME_WITH_TYPE, streamConfigMap);
    PulsarConfig pulsarConfig = new PulsarConfig(streamConfig, "testId");
    Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue> metadataFieldsToExtract =
        pulsarConfig.getMetadataFields();
    Assert.assertEquals(metadataFieldsToExtract.size(), 0);
  }