S3SchemaUriResolver implements SchemaUriResolver { @Override public URI resolve(Schema schema, String road, int version) { String key = newKey(road, version); byte[] bytes = schema.toString().getBytes(Charsets.UTF_8); ObjectMetadata objectMetadata = new ObjectMetadata(); objectMetadata.setContentLength(bytes.length); if (enableServerSideEncryption) { objectMetadata.setSSEAlgorithm(ObjectMetadata.AES_256_SERVER_SIDE_ENCRYPTION); } try (InputStream in = new ByteArrayInputStream(bytes)) { Upload upload = transferManager.upload(bucket, key, in, objectMetadata); upload.waitForCompletion(); } catch (IOException e) { throw new MetaStoreException(String.format("Error closing schema stream: bucket='%s', key='%s'", bucket, key), e); } catch (AmazonClientException | InterruptedException e) { throw new MetaStoreException(String.format("Error uploading schema: bucket='%s', key='%s'", bucket, key), e); } return newS3Uri(key); } S3SchemaUriResolver(AmazonS3 s3Client, String bucket, String keyPrefix, boolean enableServerSideEncryption); @VisibleForTesting  S3SchemaUriResolver(
      TransferManager transferManager,
      String uriFormat,
      String bucket,
      String keyPrefix,
      boolean enableServerSideEncryption); @Override URI resolve(Schema schema, String road, int version);  }
@Test public void resolve() throws Exception { when(transferManager.upload(anyString(), anyString(), any(InputStream.class), any(ObjectMetadata.class))) .thenReturn(upload); URI schemaUri = uriResolver.resolve(SCHEMA, ROAD, VERSION); verify(transferManager).upload(eq(BUCKET), eq("prefix/roads/road/schemas/1/road_v1.avsc"), uploadCaptor.capture(), metadataCaptor.capture()); verify(upload).waitForCompletion(); InputStream uploadStream = uploadCaptor.getValue(); ByteArrayOutputStream out = new ByteArrayOutputStream(); IOUtils.copy(uploadStream, out); String schemaString = new String(out.toByteArray(), Charsets.UTF_8); assertThat(schemaString, is(SCHEMA.toString())); ObjectMetadata objectMetadata = metadataCaptor.getValue(); assertThat((int) objectMetadata.getContentLength(), is(out.toByteArray().length)); assertThat(schemaUri, is(new URI("s3: }
@Test(expected = MetaStoreException.class) public void resolveAmazonServiceExceptionOnUpload() throws Exception { when(transferManager.upload(anyString(), anyString(), any(InputStream.class), any(ObjectMetadata.class))) .thenThrow(AmazonServiceException.class); uriResolver.resolve(SCHEMA, ROAD, VERSION); }
@Test(expected = MetaStoreException.class) public void resolveAmazonClientExceptionOnUpload() throws Exception { when(transferManager.upload(anyString(), anyString(), any(InputStream.class), any(ObjectMetadata.class))) .thenThrow(AmazonClientException.class); uriResolver.resolve(SCHEMA, ROAD, VERSION); }
@Test(expected = MetaStoreException.class) public void resolveIOExceptionOnUpload() throws Exception { when(transferManager.upload(anyString(), anyString(), any(InputStream.class), any(ObjectMetadata.class))) .thenThrow(IOException.class); uriResolver.resolve(SCHEMA, ROAD, VERSION); }
@Test(expected = MetaStoreException.class) public void resolveInterruptedExceptionOnUpload() throws Exception { when(transferManager.upload(anyString(), anyString(), any(InputStream.class), any(ObjectMetadata.class))) .thenThrow(InterruptedException.class); uriResolver.resolve(SCHEMA, ROAD, VERSION); }