private void writeRateLimiterConfig(Config jobConfig, Map<String, String> tableConfig) {
    if (!tagCreditsMap.isEmpty()) {
      RateLimiter defaultRateLimiter;
      try {
        @SuppressWarnings("unchecked")
        Class<? extends RateLimiter> clazz = (Class<? extends RateLimiter>) Class.forName(DEFAULT_RATE_LIMITER_CLASS_NAME);
        Constructor<? extends RateLimiter> ctor = clazz.getConstructor(Map.class);
        defaultRateLimiter = ctor.newInstance(tagCreditsMap);
      } catch (Exception ex) {
        throw new SamzaException("Failed to create default rate limiter", ex);
      }
      addTableConfig(RATE_LIMITER, SerdeUtils.serialize("rate limiter", defaultRateLimiter), tableConfig);
      if (defaultRateLimiter instanceof TablePart) {
        addTablePartConfig(RATE_LIMITER, (TablePart) defaultRateLimiter, jobConfig, tableConfig);
      }
    } else if (rateLimiter != null) {
      addTableConfig(RATE_LIMITER, SerdeUtils.serialize("rate limiter", rateLimiter), tableConfig);
      if (rateLimiter instanceof TablePart) {
        addTablePartConfig(RATE_LIMITER, (TablePart) rateLimiter, jobConfig, tableConfig);
      }
    }
    //Write table api read/write rate limit
    if (this.enableReadRateLimiter && tagCreditsMap.containsKey(RL_READ_TAG)) {
      addTableConfig(READ_CREDITS, String.valueOf(tagCreditsMap.get(RL_READ_TAG)), tableConfig);
    }
    if (this.enableWriteRateLimiter && tagCreditsMap.containsKey(RL_WRITE_TAG)) {
      addTableConfig(WRITE_CREDITS, String.valueOf(tagCreditsMap.get(RL_WRITE_TAG)), tableConfig);
    }
  }
Specifies a batch provider inorder to batch Table operations.
samza-api/src/main/java/org/apache/samza/table/descriptors/RemoteTableDescriptor.java