@Test public void testEncryptSegmentIfNeeded_uploadedSegmentIsEncrypted() { boolean uploadedSegmentIsEncrypted = true; String crypterClassNameInTableConfig = "NoOpPinotCrypter"; String crypterClassNameUsedInUploadedSegment = "NoOpPinotCrypter"; Pair<String, File> encryptionInfo = _resource .encryptSegmentIfNeeded(_decryptedFile, _encryptedFile, uploadedSegmentIsEncrypted, crypterClassNameUsedInUploadedSegment, crypterClassNameInTableConfig, SEGMENT_NAME, TABLE_NAME); assertEquals("NoOpPinotCrypter", encryptionInfo.getLeft()); assertEquals(_encryptedFile, encryptionInfo.getRight()); }
Pair<String, File> encryptSegmentIfNeeded(File tempDecryptedFile, File tempEncryptedFile, boolean isUploadedSegmentEncrypted, String crypterUsedInUploadedSegment, String crypterClassNameInTableConfig, String segmentName, String tableName) { boolean segmentNeedsEncryption = !Strings.isNullOrEmpty(crypterClassNameInTableConfig); File finalSegmentFile = (isUploadedSegmentEncrypted || segmentNeedsEncryption) ? tempEncryptedFile : tempDecryptedFile; String crypterClassName = Strings.isNullOrEmpty(crypterClassNameInTableConfig) ? crypterUsedInUploadedSegment : crypterClassNameInTableConfig; ImmutablePair<String, File> out = ImmutablePair.of(crypterClassName, finalSegmentFile); if (!segmentNeedsEncryption) { return out; } if (isUploadedSegmentEncrypted && !crypterClassNameInTableConfig.equals(crypterUsedInUploadedSegment)) { throw new ControllerApplicationException(LOGGER, String.format( "Uploaded segment is encrypted with '%s' while table config requires '%s' as crypter " + "(segment name = '%s', table name = '%s').", crypterUsedInUploadedSegment, crypterClassNameInTableConfig, segmentName, tableName), Response.Status.INTERNAL_SERVER_ERROR); } PinotCrypter pinotCrypter = PinotCrypterFactory.create(crypterClassNameInTableConfig); LOGGER.info("Using crypter class '{}' for encrypting '{}' to '{}' (segment name = '{}', table name = '{}').", crypterClassNameInTableConfig, tempDecryptedFile, tempEncryptedFile, segmentName, tableName); pinotCrypter.encrypt(tempDecryptedFile, tempEncryptedFile); return out; }
PinotSegmentUploadDownloadRestletResource { Pair<String, File> encryptSegmentIfNeeded(File tempDecryptedFile, File tempEncryptedFile, boolean isUploadedSegmentEncrypted, String crypterUsedInUploadedSegment, String crypterClassNameInTableConfig, String segmentName, String tableName) { boolean segmentNeedsEncryption = !Strings.isNullOrEmpty(crypterClassNameInTableConfig); File finalSegmentFile = (isUploadedSegmentEncrypted || segmentNeedsEncryption) ? tempEncryptedFile : tempDecryptedFile; String crypterClassName = Strings.isNullOrEmpty(crypterClassNameInTableConfig) ? crypterUsedInUploadedSegment : crypterClassNameInTableConfig; ImmutablePair<String, File> out = ImmutablePair.of(crypterClassName, finalSegmentFile); if (!segmentNeedsEncryption) { return out; } if (isUploadedSegmentEncrypted && !crypterClassNameInTableConfig.equals(crypterUsedInUploadedSegment)) { throw new ControllerApplicationException(LOGGER, String.format( "Uploaded segment is encrypted with '%s' while table config requires '%s' as crypter " + "(segment name = '%s', table name = '%s').", crypterUsedInUploadedSegment, crypterClassNameInTableConfig, segmentName, tableName), Response.Status.INTERNAL_SERVER_ERROR); } PinotCrypter pinotCrypter = PinotCrypterFactory.create(crypterClassNameInTableConfig); LOGGER.info("Using crypter class '{}' for encrypting '{}' to '{}' (segment name = '{}', table name = '{}').", crypterClassNameInTableConfig, tempDecryptedFile, tempEncryptedFile, segmentName, tableName); pinotCrypter.encrypt(tempDecryptedFile, tempEncryptedFile); return out; } }
PinotSegmentUploadDownloadRestletResource { Pair<String, File> encryptSegmentIfNeeded(File tempDecryptedFile, File tempEncryptedFile, boolean isUploadedSegmentEncrypted, String crypterUsedInUploadedSegment, String crypterClassNameInTableConfig, String segmentName, String tableName) { boolean segmentNeedsEncryption = !Strings.isNullOrEmpty(crypterClassNameInTableConfig); File finalSegmentFile = (isUploadedSegmentEncrypted || segmentNeedsEncryption) ? tempEncryptedFile : tempDecryptedFile; String crypterClassName = Strings.isNullOrEmpty(crypterClassNameInTableConfig) ? crypterUsedInUploadedSegment : crypterClassNameInTableConfig; ImmutablePair<String, File> out = ImmutablePair.of(crypterClassName, finalSegmentFile); if (!segmentNeedsEncryption) { return out; } if (isUploadedSegmentEncrypted && !crypterClassNameInTableConfig.equals(crypterUsedInUploadedSegment)) { throw new ControllerApplicationException(LOGGER, String.format( "Uploaded segment is encrypted with '%s' while table config requires '%s' as crypter " + "(segment name = '%s', table name = '%s').", crypterUsedInUploadedSegment, crypterClassNameInTableConfig, segmentName, tableName), Response.Status.INTERNAL_SERVER_ERROR); } PinotCrypter pinotCrypter = PinotCrypterFactory.create(crypterClassNameInTableConfig); LOGGER.info("Using crypter class '{}' for encrypting '{}' to '{}' (segment name = '{}', table name = '{}').", crypterClassNameInTableConfig, tempDecryptedFile, tempEncryptedFile, segmentName, tableName); pinotCrypter.encrypt(tempDecryptedFile, tempEncryptedFile); return out; }  }
PinotSegmentUploadDownloadRestletResource { Pair<String, File> encryptSegmentIfNeeded(File tempDecryptedFile, File tempEncryptedFile, boolean isUploadedSegmentEncrypted, String crypterUsedInUploadedSegment, String crypterClassNameInTableConfig, String segmentName, String tableName) { boolean segmentNeedsEncryption = !Strings.isNullOrEmpty(crypterClassNameInTableConfig); File finalSegmentFile = (isUploadedSegmentEncrypted || segmentNeedsEncryption) ? tempEncryptedFile : tempDecryptedFile; String crypterClassName = Strings.isNullOrEmpty(crypterClassNameInTableConfig) ? crypterUsedInUploadedSegment : crypterClassNameInTableConfig; ImmutablePair<String, File> out = ImmutablePair.of(crypterClassName, finalSegmentFile); if (!segmentNeedsEncryption) { return out; } if (isUploadedSegmentEncrypted && !crypterClassNameInTableConfig.equals(crypterUsedInUploadedSegment)) { throw new ControllerApplicationException(LOGGER, String.format( "Uploaded segment is encrypted with '%s' while table config requires '%s' as crypter " + "(segment name = '%s', table name = '%s').", crypterUsedInUploadedSegment, crypterClassNameInTableConfig, segmentName, tableName), Response.Status.INTERNAL_SERVER_ERROR); } PinotCrypter pinotCrypter = PinotCrypterFactory.create(crypterClassNameInTableConfig); LOGGER.info("Using crypter class '{}' for encrypting '{}' to '{}' (segment name = '{}', table name = '{}').", crypterClassNameInTableConfig, tempDecryptedFile, tempEncryptedFile, segmentName, tableName); pinotCrypter.encrypt(tempDecryptedFile, tempEncryptedFile); return out; }  @GET @Produces(MediaType.APPLICATION_OCTET_STREAM) @Path("/segments/{tableName}/{segmentName}") @ApiOperation(value = "Download a segment", notes = "Download a segment") Response downloadSegment(
      @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
      @ApiParam(value = "Name of the segment", required = true) @PathParam("segmentName") @Encoded String segmentName,
      @Context HttpHeaders httpHeaders); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) @Path("/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as json") // We use this endpoint with URI upload because a request sent with the multipart content type will reject the POST // request if a multipart object is not sent. This endpoint does not move the segment to its final location; // it keeps it at the downloadURI header that is set. We will not support this endpoint going forward. void uploadSegmentAsJson(String segmentJsonStr,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as binary") // For the multipart endpoint, we will always move segment to final location regardless of the segment endpoint. void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) @Path("/v2/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as json") // We use this endpoint with URI upload because a request sent with the multipart content type will reject the POST // request if a multipart object is not sent. This endpoint is recommended for use. It differs from the first // endpoint in how it moves the segment to a Pinot-determined final directory. void uploadSegmentAsJsonV2(String segmentJsonStr,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/v2/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as binary") // This behavior does not differ from v1 of the same endpoint. void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @Path("segments/{tableName}/startReplaceSegments") @Produces(MediaType.APPLICATION_JSON) @ApiOperation(value = "Start to replace segments", notes = "Start to replace segments") Response startReplaceSegments(
      @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
      @ApiParam(value = "OFFLINE|REALTIME") @QueryParam("type") String tableTypeStr,
      StartReplaceSegmentsRequest startReplaceSegmentsRequest); @POST @Path("segments/{tableName}/endReplaceSegments") @Produces(MediaType.APPLICATION_JSON) @ApiOperation(value = "End to replace segments", notes = "End to replace segments") Response endReplaceSegments(
      @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
      @ApiParam(value = "OFFLINE|REALTIME") @QueryParam("type") String tableTypeStr,
      @ApiParam(value = "Segment lineage entry id returned by startReplaceSegments API") @QueryParam("segmentLineageEntryId") String segmentLineageEntryId); static boolean validateMultiPart(Map<String, List<FormDataBodyPart>> map, String segmentName); }
PinotSegmentUploadDownloadRestletResource { Pair<String, File> encryptSegmentIfNeeded(File tempDecryptedFile, File tempEncryptedFile, boolean isUploadedSegmentEncrypted, String crypterUsedInUploadedSegment, String crypterClassNameInTableConfig, String segmentName, String tableName) { boolean segmentNeedsEncryption = !Strings.isNullOrEmpty(crypterClassNameInTableConfig); File finalSegmentFile = (isUploadedSegmentEncrypted || segmentNeedsEncryption) ? tempEncryptedFile : tempDecryptedFile; String crypterClassName = Strings.isNullOrEmpty(crypterClassNameInTableConfig) ? crypterUsedInUploadedSegment : crypterClassNameInTableConfig; ImmutablePair<String, File> out = ImmutablePair.of(crypterClassName, finalSegmentFile); if (!segmentNeedsEncryption) { return out; } if (isUploadedSegmentEncrypted && !crypterClassNameInTableConfig.equals(crypterUsedInUploadedSegment)) { throw new ControllerApplicationException(LOGGER, String.format( "Uploaded segment is encrypted with '%s' while table config requires '%s' as crypter " + "(segment name = '%s', table name = '%s').", crypterUsedInUploadedSegment, crypterClassNameInTableConfig, segmentName, tableName), Response.Status.INTERNAL_SERVER_ERROR); } PinotCrypter pinotCrypter = PinotCrypterFactory.create(crypterClassNameInTableConfig); LOGGER.info("Using crypter class '{}' for encrypting '{}' to '{}' (segment name = '{}', table name = '{}').", crypterClassNameInTableConfig, tempDecryptedFile, tempEncryptedFile, segmentName, tableName); pinotCrypter.encrypt(tempDecryptedFile, tempEncryptedFile); return out; }  @GET @Produces(MediaType.APPLICATION_OCTET_STREAM) @Path("/segments/{tableName}/{segmentName}") @ApiOperation(value = "Download a segment", notes = "Download a segment") Response downloadSegment(
      @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
      @ApiParam(value = "Name of the segment", required = true) @PathParam("segmentName") @Encoded String segmentName,
      @Context HttpHeaders httpHeaders); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) @Path("/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as json") // We use this endpoint with URI upload because a request sent with the multipart content type will reject the POST // request if a multipart object is not sent. This endpoint does not move the segment to its final location; // it keeps it at the downloadURI header that is set. We will not support this endpoint going forward. void uploadSegmentAsJson(String segmentJsonStr,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as binary") // For the multipart endpoint, we will always move segment to final location regardless of the segment endpoint. void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) @Path("/v2/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as json") // We use this endpoint with URI upload because a request sent with the multipart content type will reject the POST // request if a multipart object is not sent. This endpoint is recommended for use. It differs from the first // endpoint in how it moves the segment to a Pinot-determined final directory. void uploadSegmentAsJsonV2(String segmentJsonStr,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @ManagedAsync @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/v2/segments") @ApiOperation(value = "Upload a segment", notes = "Upload a segment as binary") // This behavior does not differ from v1 of the same endpoint. void uploadSegmentAsMultiPartV2(FormDataMultiPart multiPart,
      @ApiParam(value = "Name of the table") @QueryParam(FileUploadDownloadClient.QueryParameters.TABLE_NAME) String tableName,
      @ApiParam(value = "Whether to enable parallel push protection") @DefaultValue("false") @QueryParam(FileUploadDownloadClient.QueryParameters.ENABLE_PARALLEL_PUSH_PROTECTION) boolean enableParallelPushProtection,
      @Context HttpHeaders headers, @Context Request request, @Suspended final AsyncResponse asyncResponse); @POST @Path("segments/{tableName}/startReplaceSegments") @Produces(MediaType.APPLICATION_JSON) @ApiOperation(value = "Start to replace segments", notes = "Start to replace segments") Response startReplaceSegments(
      @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
      @ApiParam(value = "OFFLINE|REALTIME") @QueryParam("type") String tableTypeStr,
      StartReplaceSegmentsRequest startReplaceSegmentsRequest); @POST @Path("segments/{tableName}/endReplaceSegments") @Produces(MediaType.APPLICATION_JSON) @ApiOperation(value = "End to replace segments", notes = "End to replace segments") Response endReplaceSegments(
      @ApiParam(value = "Name of the table", required = true) @PathParam("tableName") String tableName,
      @ApiParam(value = "OFFLINE|REALTIME") @QueryParam("type") String tableTypeStr,
      @ApiParam(value = "Segment lineage entry id returned by startReplaceSegments API") @QueryParam("segmentLineageEntryId") String segmentLineageEntryId); static boolean validateMultiPart(Map<String, List<FormDataBodyPart>> map, String segmentName);  }