static int validatePubsubMessageSize(PubsubMessage message, int maxPublishBatchSize)
      throws SizeLimitExceededException {
    int payloadSize = message.getPayload().length;
    if (payloadSize > PUBSUB_MESSAGE_DATA_MAX_BYTES) {
      throw new SizeLimitExceededException(
          "Pubsub message data field of length "
              + payloadSize
              + " exceeds maximum of "
              + PUBSUB_MESSAGE_DATA_MAX_BYTES
              + " bytes. See https://cloud.google.com/pubsub/quotas#resource_limits");
    }
    int totalSize = payloadSize;

    @Nullable Map<String, String> attributes = message.getAttributeMap();
    if (attributes != null) {
      if (attributes.size() > PUBSUB_MESSAGE_MAX_ATTRIBUTES) {
        throw new SizeLimitExceededException(
            "Pubsub message contains "
                + attributes.size()
                + " attributes which exceeds the maximum of "
                + PUBSUB_MESSAGE_MAX_ATTRIBUTES
                + ". See https://cloud.google.com/pubsub/quotas#resource_limits");
      }

      // Consider attribute encoding overhead, so it doesn't go over the request limits
      totalSize += attributes.size() * PUBSUB_MESSAGE_ATTRIBUTE_ENCODE_ADDITIONAL_BYTES;

      for (Map.Entry<String, String> attribute : attributes.entrySet()) {
        String key = attribute.getKey();
        int keySize = key.getBytes(StandardCharsets.UTF_8).length;
        if (keySize > PUBSUB_MESSAGE_ATTRIBUTE_MAX_KEY_BYTES) {
          throw new SizeLimitExceededException(
              "Pubsub message attribute key '"
                  + key
                  + "' exceeds the maximum of "
                  + PUBSUB_MESSAGE_ATTRIBUTE_MAX_KEY_BYTES
                  + " bytes. See https://cloud.google.com/pubsub/quotas#resource_limits");
        }
        totalSize += keySize;

        String value = attribute.getValue();
        int valueSize = value.getBytes(StandardCharsets.UTF_8).length;
        if (valueSize > PUBSUB_MESSAGE_ATTRIBUTE_MAX_VALUE_BYTES) {
          throw new SizeLimitExceededException(
              "Pubsub message attribute value for key '"
                  + key
                  + "' starting with '"
                  + value.substring(0, Math.min(256, value.length()))
                  + "' exceeds the maximum of "
                  + PUBSUB_MESSAGE_ATTRIBUTE_MAX_VALUE_BYTES
                  + " bytes. See https://cloud.google.com/pubsub/quotas#resource_limits");
        }
        totalSize += valueSize;
      }
    }

    if (totalSize > maxPublishBatchSize) {
      throw new SizeLimitExceededException(
          "Pubsub message of length "
              + totalSize
              + " exceeds maximum of "
              + maxPublishBatchSize
              + " bytes, when considering the payload and attributes. "
              + "See https://cloud.google.com/pubsub/quotas#resource_limits");
    }
    return totalSize;
  }
@Test
  public void testValidatePubsubMessageSizeOnlyPayload() throws SizeLimitExceededException {
    byte[] data = new byte[1024];
    PubsubMessage message = new PubsubMessage(data, null);

    int messageSize =
        PreparePubsubWriteDoFn.validatePubsubMessageSize(message, PUBSUB_MESSAGE_MAX_TOTAL_SIZE);

    assertEquals(data.length, messageSize);
  }