S3 offloader doesn't allow block size < 5MB (apache#1809)

S3 doesn't allow multipart upload to be used with a block size less
than 5MB. So we shouldn't allow our offloader to be configured with a
value lower than 5MB.

Master Issue: apache#1511

conf/broker.conf

@@ -496,7 +496,7 @@ s3ManagedLedgerOffloadBucket=
 # For Amazon S3 ledger offload, Alternative endpoint to connect to (useful for testing)
 s3ManagedLedgerOffloadServiceEndpoint=
 
-# For Amazon S3 ledger offload, Max block size in bytes.
+# For Amazon S3 ledger offload, Max block size in bytes. (64MB by default, 5MB minimum)
 s3ManagedLedgerOffloadMaxBlockSizeInBytes=67108864
 
 # For Amazon S3 ledger offload, Read buffer size in bytes (1MB by default)

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -489,7 +489,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
     private String s3ManagedLedgerOffloadServiceEndpoint = null;
 
     // For Amazon S3 ledger offload, Max block size in bytes.
-    private int s3ManagedLedgerOffloadMaxBlockSizeInBytes = 64 * 1024 * 1024;
+    @FieldContext(minValue = 5242880) // 5MB
+    private int s3ManagedLedgerOffloadMaxBlockSizeInBytes = 64 * 1024 * 1024; // 64MB
 
     // For Amazon S3 ledger offload, Read buffer size in bytes.
     @FieldContext(minValue = 1024)

pulsar-broker/src/main/java/org/apache/pulsar/broker/s3offload/S3ManagedLedgerOffloader.java

@@ -76,6 +76,9 @@ public static S3ManagedLedgerOffloader create(ServiceConfiguration conf,
         if (Strings.isNullOrEmpty(bucket)) {
             throw new PulsarServerException("s3ManagedLedgerOffloadBucket cannot be empty if s3 offload enabled");
         }
+        if (maxBlockSize < 5*1024*1024) {
+            throw new PulsarServerException("s3ManagedLedgerOffloadMaxBlockSizeInBytes cannot be less than 5MB");
+        }
 
         AmazonS3ClientBuilder builder = AmazonS3ClientBuilder.standard();
         if (!Strings.isNullOrEmpty(endpoint)) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/s3offload/S3ManagedLedgerOffloaderTest.java

@@ -159,6 +159,22 @@ public void testNoBucketConfigured() throws Exception {
         }
     }
 
+    @Test
+    public void testSmallBlockSizeConfigured() throws Exception {
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setManagedLedgerOffloadDriver(S3ManagedLedgerOffloader.DRIVER_NAME);
+        conf.setS3ManagedLedgerOffloadRegion("eu-west-1");
+        conf.setS3ManagedLedgerOffloadBucket(BUCKET);
+        conf.setS3ManagedLedgerOffloadMaxBlockSizeInBytes(1024);
+
+        try {
+            S3ManagedLedgerOffloader.create(conf, scheduler);
+            Assert.fail("Should have thrown exception");
+        } catch (PulsarServerException pse) {
+            // correct
+        }
+    }
+
     @Test
     public void testOffloadAndRead() throws Exception {
         ReadHandle toWrite = buildReadHandle(DEFAULT_BLOCK_SIZE, 3);