public Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue> getMetadataFields() {
    return _metadataFields;
  }
@Test
  public void testParsingMetadataConfigWithConfiguredFields() {
    Map<String, String> streamConfigMap = getCommonStreamConfigMap();
    streamConfigMap.put(
        StreamConfigProperties.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.METADATA_POPULATE), "true");
    streamConfigMap.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE, PulsarConfig.METADATA_FIELDS),
        "messageId,messageIdBytes, publishTime, eventTime, key, topicName, ");
    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(), 6);
    Assert.assertTrue(metadataFieldsToExtract.containsAll(
        ImmutableList.of(PulsarStreamMessageMetadata.PulsarMessageMetadataValue.MESSAGE_ID,
            PulsarStreamMessageMetadata.PulsarMessageMetadataValue.MESSAGE_ID_BYTES_B64,
            PulsarStreamMessageMetadata.PulsarMessageMetadataValue.PUBLISH_TIME,
            PulsarStreamMessageMetadata.PulsarMessageMetadataValue.EVENT_TIME,
            PulsarStreamMessageMetadata.PulsarMessageMetadataValue.MESSAGE_KEY,
            PulsarStreamMessageMetadata.PulsarMessageMetadataValue.TOPIC_NAME)));
  }