fix storage class npe

s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentMover.java

@@ -24,6 +24,7 @@ import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.inject.Inject;
+import com.metamx.common.ISE;
 import com.metamx.common.MapUtils;
 import com.metamx.common.logger.Logger;
 import io.druid.segment.loading.DataSegmentMover;
@@ -120,23 +121,30 @@ public class S3DataSegmentMover implements DataSegmentMover
               if (s3Client.isObjectInBucket(s3Bucket, s3Path)) {
                 if (s3Bucket.equals(targetS3Bucket) && s3Path.equals(targetS3Path)) {
                   log.info("No need to move file[s3://%s/%s] onto itself", s3Bucket, s3Path);
-                } else if (s3Client.getObjectDetails(s3Bucket, s3Path)
-                                   .getStorageClass()
-                                   .equals(S3Object.STORAGE_CLASS_GLACIER)) {
-                  log.warn("Cannot move file[s3://%s/%s] of storage class glacier.");
                 } else {
-                  log.info(
-                      "Moving file[s3://%s/%s] to [s3://%s/%s]",
-                      s3Bucket,
-                      s3Path,
-                      targetS3Bucket,
-                      targetS3Path
-                  );
-                  final S3Object target = new S3Object(targetS3Path);
-                  if(!config.getDisableAcl()) {
-                    target.setAcl(GSAccessControlList.REST_CANNED_BUCKET_OWNER_FULL_CONTROL);
+                  final S3Object[] list = s3Client.listObjects(s3Bucket, s3Path, "");
+                  if (list.length == 0) {
+                    // should never happen
+                    throw new ISE("Unable to list object [s3://%s/%s]", s3Bucket, s3Path);
+                  }
+                  final S3Object s3Object = list[0];
+                  if (s3Object.getStorageClass() != null &&
+                      s3Object.getStorageClass().equals(S3Object.STORAGE_CLASS_GLACIER)) {
+                    log.warn("Cannot move file[s3://%s/%s] of storage class glacier, skipping.");
+                  } else {
+                    log.info(
+                        "Moving file[s3://%s/%s] to [s3://%s/%s]",
+                        s3Bucket,
+                        s3Path,
+                        targetS3Bucket,
+                        targetS3Path
+                    );
+                    final S3Object target = new S3Object(targetS3Path);
+                    if (!config.getDisableAcl()) {
+                      target.setAcl(GSAccessControlList.REST_CANNED_BUCKET_OWNER_FULL_CONTROL);
+                    }
+                    s3Client.moveObject(s3Bucket, s3Path, targetS3Bucket, target, false);
                   }
-                  s3Client.moveObject(s3Bucket, s3Path, targetS3Bucket, target, false);
                 }
               } else {
                 // ensure object exists in target location