@Override
    public StorageObject upload(final Path file, Local local, final BandwidthThrottle throttle, final StreamListener listener,
                                final TransferStatus status, final ConnectionCallback prompt) throws BackgroundException {
        if(this.threshold(status)) {
            try {
                return new S3MultipartUploadService(session, writer, acl).upload(file, local, throttle, listener, status, prompt);
            }
            catch(NotfoundException | InteroperabilityException e) {
                log.warn(String.format("Failure %s using multipart upload. Fallback to single upload.", e));
                status.append(false);
                try {
                    return new S3SingleUploadService(session, writer).upload(file, local, throttle, listener, status, prompt);
                }
                catch(BackgroundException f) {
                    log.warn(String.format("Failure %s using single upload. Throw original multipart failure %s", e, e));
                    throw e;
                }
            }
        }
        // Use single upload service
        return new S3SingleUploadService(session, writer).upload(file, local, throttle, listener, status, prompt);
    }
@Test
    public void testUploadSinglePartEuCentral() throws Exception {
        final S3ThresholdUploadService service = new S3ThresholdUploadService(session, new S3AccessControlListFeature(session), 5 * 1024L);
        final Path container = new Path("test-eu-central-1-cyberduck", EnumSet.of(Path.Type.directory, Path.Type.volume));
        final String name = UUID.randomUUID().toString();
        final Path test = new Path(container, name, EnumSet.of(Path.Type.file));
        final Local local = new Local(System.getProperty("java.io.tmpdir"), name);
        final byte[] random = RandomUtils.nextBytes(1000);
        IOUtils.write(random, local.getOutputStream(false));
        final TransferStatus status = new TransferStatus();
        status.setLength(random.length);
        status.setStorageClass(S3Object.STORAGE_CLASS_REDUCED_REDUNDANCY);
        final BytecountStreamListener count = new BytecountStreamListener();
        service.upload(test, local, new BandwidthThrottle(BandwidthThrottle.UNLIMITED),
                count, status, new DisabledLoginCallback());
        assertEquals(random.length, count.getSent(), 0L);
        assertTrue(status.isComplete());
        assertTrue(new S3FindFeature(session, new S3AccessControlListFeature(session)).find(test));
        final PathAttributes attributes = new S3AttributesFinderFeature(session, new S3AccessControlListFeature(session)).find(test);
        assertEquals(random.length, attributes.getSize(), 0L);
        assertEquals(S3Object.STORAGE_CLASS_REDUCED_REDUNDANCY, new S3StorageClassFeature(session, new S3AccessControlListFeature(session)).getClass(test));
        new S3DefaultDeleteFeature(session).delete(Collections.singletonList(test), new DisabledLoginCallback(), new Delete.DisabledCallback());
        local.delete();
    }