@VisibleForTesting
  static StreamMessageMetadata extractMessageMetadata(Message<byte[]> message, PulsarConfig config) {
    long recordIngestionTimeMs = message.getBrokerPublishTime().orElse(message.getPublishTime());
    MessageId messageId = message.getMessageId();
    MessageIdStreamOffset offset = new MessageIdStreamOffset(messageId);
    MessageIdStreamOffset nextOffset = new MessageIdStreamOffset(getNextMessageId(messageId));
    StreamMessageMetadata.Builder builder =
        new StreamMessageMetadata.Builder().setRecordIngestionTimeMs(recordIngestionTimeMs)
            .setOffset(offset, nextOffset);
    if (config.isPopulateMetadata()) {
      Map<String, String> properties = message.getProperties();
      if (!properties.isEmpty()) {
        GenericRow header = new GenericRow();
        for (Map.Entry<String, String> entry : properties.entrySet()) {
          header.putValue(entry.getKey(), entry.getValue());
        }
        builder.setHeaders(header);
      }
      Set<PulsarStreamMessageMetadata.PulsarMessageMetadataValue> metadataFields = config.getMetadataFields();
      if (!metadataFields.isEmpty()) {
        Map<String, String> metadataMap = Maps.newHashMapWithExpectedSize(metadataFields.size());
        for (PulsarStreamMessageMetadata.PulsarMessageMetadataValue metadataField : metadataFields) {
          populateMetadataField(message, metadataField, metadataMap);
        }
        builder.setMetadata(metadataMap);
      }
    }
    return builder.build();
  }
@Test
  public void testExtractProperty()
      throws Exception {
    PulsarConfig config = mock(PulsarConfig.class);
    when(config.isPopulateMetadata()).thenReturn(true);
    when(config.getMetadataFields()).thenReturn(Set.of(MESSAGE_ID, MESSAGE_ID_BYTES_B64, MESSAGE_KEY));
    DummyPulsarMessage pulsarMessage =
        new DummyPulsarMessage("key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8));
    pulsarMessage.getProperties().put("test_key", "test_value");
    pulsarMessage.getProperties().put("test_key2", "2");
    StreamMessageMetadata metadata = PulsarUtils.extractMessageMetadata(pulsarMessage, config);
    GenericRow headers = metadata.getHeaders();
    assertNotNull(headers);
    assertEquals(headers.getValue("test_key"), "test_value");
    assertEquals(headers.getValue("test_key2"), "2");
    Map<String, String> recordMetadata = metadata.getRecordMetadata();
    assertNotNull(recordMetadata);
    assertEquals(recordMetadata.get(MESSAGE_KEY.getKey()), "key");
    assertEquals(recordMetadata.get(MESSAGE_ID.getKey()), pulsarMessage.getMessageId().toString());

    byte[] messageIdBytes = Base64.decode(recordMetadata.get(MESSAGE_ID_BYTES_B64.getKey()));
    MessageId messageId = MessageId.fromByteArray(messageIdBytes);
    assertEquals(MessageId.earliest, messageId);
  }