@Test public void testProcessUdfWithInnerClassFqdnName() throws Exception{ String fnName = "FooPredicateGTZ"; String fqdnName = "FooInnerclassTestUDFs$FooPredicateGTZ"; String canonicalName = "FooInnerclassTestUDFs.FooPredicateGTZ.FooAdder"; String className = getFullClassName(fqdnName); UDF udf = newTestUDF(fnName, className); InputStream inputStream = new FileInputStream(udf.getJarStoragePath()); catalogResource.processUdf(inputStream, udf, true, false); Assert.assertEquals(getFullClassName(fqdnName), udf.getClassName()); }
@VisibleForTesting void processUdf(InputStream inputStream, UDF udf, boolean checkDuplicate, boolean builtin) throws Exception { if (builtin) { udf.setDigest("builtin"); udf.setJarStoragePath("builtin"); checkDuplicate(udf); } else { MessageDigest md = MessageDigest.getInstance("SHA-256"); File tmpFile; try (DigestInputStream dis = new DigestInputStream(inputStream, md)) { tmpFile = FileUtil.writeInputStreamToTempFile(dis, ".jar"); } Map<String, Class<?>> udfs = StreamCatalogService.loadUdfsFromJar(tmpFile); validateUDF(udfs.keySet(), udf, checkDuplicate); updateTypeInfo(udf, udfs.get(udf.getClassName())); String digest = Hex.encodeHexString(md.digest()); LOG.debug("Digest: {}", digest); udf.setDigest(digest); String jarPath = getExistingJarPath(digest).orElseGet(() -> uploadJar(tmpFile, udf.getName())); if (!fileStorage.exists(jarPath)) { String msg = String.format("The jar path '%s' does not exist. " + "You may have to reset the db and run bootstrap again.", jarPath); LOG.error(msg); throw new RuntimeException(msg); } udf.setJarStoragePath(jarPath); } }
UDFCatalogResource { @VisibleForTesting void processUdf(InputStream inputStream, UDF udf, boolean checkDuplicate, boolean builtin) throws Exception { if (builtin) { udf.setDigest("builtin"); udf.setJarStoragePath("builtin"); checkDuplicate(udf); } else { MessageDigest md = MessageDigest.getInstance("SHA-256"); File tmpFile; try (DigestInputStream dis = new DigestInputStream(inputStream, md)) { tmpFile = FileUtil.writeInputStreamToTempFile(dis, ".jar"); } Map<String, Class<?>> udfs = StreamCatalogService.loadUdfsFromJar(tmpFile); validateUDF(udfs.keySet(), udf, checkDuplicate); updateTypeInfo(udf, udfs.get(udf.getClassName())); String digest = Hex.encodeHexString(md.digest()); LOG.debug("Digest: {}", digest); udf.setDigest(digest); String jarPath = getExistingJarPath(digest).orElseGet(() -> uploadJar(tmpFile, udf.getName())); if (!fileStorage.exists(jarPath)) { String msg = String.format("The jar path '%s' does not exist. " + "You may have to reset the db and run bootstrap again.", jarPath); LOG.error(msg); throw new RuntimeException(msg); } udf.setJarStoragePath(jarPath); } } }
UDFCatalogResource { @VisibleForTesting void processUdf(InputStream inputStream, UDF udf, boolean checkDuplicate, boolean builtin) throws Exception { if (builtin) { udf.setDigest("builtin"); udf.setJarStoragePath("builtin"); checkDuplicate(udf); } else { MessageDigest md = MessageDigest.getInstance("SHA-256"); File tmpFile; try (DigestInputStream dis = new DigestInputStream(inputStream, md)) { tmpFile = FileUtil.writeInputStreamToTempFile(dis, ".jar"); } Map<String, Class<?>> udfs = StreamCatalogService.loadUdfsFromJar(tmpFile); validateUDF(udfs.keySet(), udf, checkDuplicate); updateTypeInfo(udf, udfs.get(udf.getClassName())); String digest = Hex.encodeHexString(md.digest()); LOG.debug("Digest: {}", digest); udf.setDigest(digest); String jarPath = getExistingJarPath(digest).orElseGet(() -> uploadJar(tmpFile, udf.getName())); if (!fileStorage.exists(jarPath)) { String msg = String.format("The jar path '%s' does not exist. " + "You may have to reset the db and run bootstrap again.", jarPath); LOG.error(msg); throw new RuntimeException(msg); } udf.setJarStoragePath(jarPath); } } UDFCatalogResource(StreamlineAuthorizer authorizer, StreamCatalogService catalogService, FileStorage fileStorage); }
UDFCatalogResource { @VisibleForTesting void processUdf(InputStream inputStream, UDF udf, boolean checkDuplicate, boolean builtin) throws Exception { if (builtin) { udf.setDigest("builtin"); udf.setJarStoragePath("builtin"); checkDuplicate(udf); } else { MessageDigest md = MessageDigest.getInstance("SHA-256"); File tmpFile; try (DigestInputStream dis = new DigestInputStream(inputStream, md)) { tmpFile = FileUtil.writeInputStreamToTempFile(dis, ".jar"); } Map<String, Class<?>> udfs = StreamCatalogService.loadUdfsFromJar(tmpFile); validateUDF(udfs.keySet(), udf, checkDuplicate); updateTypeInfo(udf, udfs.get(udf.getClassName())); String digest = Hex.encodeHexString(md.digest()); LOG.debug("Digest: {}", digest); udf.setDigest(digest); String jarPath = getExistingJarPath(digest).orElseGet(() -> uploadJar(tmpFile, udf.getName())); if (!fileStorage.exists(jarPath)) { String msg = String.format("The jar path '%s' does not exist. " + "You may have to reset the db and run bootstrap again.", jarPath); LOG.error(msg); throw new RuntimeException(msg); } udf.setJarStoragePath(jarPath); } } UDFCatalogResource(StreamlineAuthorizer authorizer, StreamCatalogService catalogService, FileStorage fileStorage); @GET @Path("/udfs") @Timed Response listUDFs(@Context UriInfo uriInfo,
                             @Context SecurityContext securityContext); @GET @Path("/udfs/{id}") @Timed Response getUDFById(@PathParam("id") Long id, @Context SecurityContext securityContext); @Timed @POST @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/udfs") Response addUDF(@FormDataParam("udfJarFile") final InputStream inputStream,
                           @FormDataParam("udfJarFile") final FormDataContentDisposition contentDispositionHeader,
                           @FormDataParam("udfConfig") final FormDataBodyPart udfConfig,
                           @FormDataParam("builtin") final boolean builtin,
                           @Context SecurityContext securityContext); @DELETE @Path("/udfs/{id}") @Timed Response removeUDF(@PathParam("id") Long id, @Context SecurityContext securityContext); @PUT @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/udfs/{id}") @Timed Response addOrUpdateUDF(@PathParam("id") Long udfId,
                                   @FormDataParam("udfJarFile") final InputStream inputStream,
                                   @FormDataParam("udfJarFile") final FormDataContentDisposition contentDispositionHeader,
                                   @FormDataParam("udfConfig") final FormDataBodyPart udfConfig,
                                   @FormDataParam("builtin") final boolean builtin,
                                   @Context SecurityContext securityContext); @Timed @GET @Produces({"application/java-archive", "application/json"}) @Path("/udfs/download/{udfId}") Response downloadUdf(@PathParam("udfId") Long udfId, @Context SecurityContext securityContext); }
UDFCatalogResource { @VisibleForTesting void processUdf(InputStream inputStream, UDF udf, boolean checkDuplicate, boolean builtin) throws Exception { if (builtin) { udf.setDigest("builtin"); udf.setJarStoragePath("builtin"); checkDuplicate(udf); } else { MessageDigest md = MessageDigest.getInstance("SHA-256"); File tmpFile; try (DigestInputStream dis = new DigestInputStream(inputStream, md)) { tmpFile = FileUtil.writeInputStreamToTempFile(dis, ".jar"); } Map<String, Class<?>> udfs = StreamCatalogService.loadUdfsFromJar(tmpFile); validateUDF(udfs.keySet(), udf, checkDuplicate); updateTypeInfo(udf, udfs.get(udf.getClassName())); String digest = Hex.encodeHexString(md.digest()); LOG.debug("Digest: {}", digest); udf.setDigest(digest); String jarPath = getExistingJarPath(digest).orElseGet(() -> uploadJar(tmpFile, udf.getName())); if (!fileStorage.exists(jarPath)) { String msg = String.format("The jar path '%s' does not exist. " + "You may have to reset the db and run bootstrap again.", jarPath); LOG.error(msg); throw new RuntimeException(msg); } udf.setJarStoragePath(jarPath); } } UDFCatalogResource(StreamlineAuthorizer authorizer, StreamCatalogService catalogService, FileStorage fileStorage); @GET @Path("/udfs") @Timed Response listUDFs(@Context UriInfo uriInfo,
                             @Context SecurityContext securityContext); @GET @Path("/udfs/{id}") @Timed Response getUDFById(@PathParam("id") Long id, @Context SecurityContext securityContext); @Timed @POST @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/udfs") Response addUDF(@FormDataParam("udfJarFile") final InputStream inputStream,
                           @FormDataParam("udfJarFile") final FormDataContentDisposition contentDispositionHeader,
                           @FormDataParam("udfConfig") final FormDataBodyPart udfConfig,
                           @FormDataParam("builtin") final boolean builtin,
                           @Context SecurityContext securityContext); @DELETE @Path("/udfs/{id}") @Timed Response removeUDF(@PathParam("id") Long id, @Context SecurityContext securityContext); @PUT @Consumes(MediaType.MULTIPART_FORM_DATA) @Path("/udfs/{id}") @Timed Response addOrUpdateUDF(@PathParam("id") Long udfId,
                                   @FormDataParam("udfJarFile") final InputStream inputStream,
                                   @FormDataParam("udfJarFile") final FormDataContentDisposition contentDispositionHeader,
                                   @FormDataParam("udfConfig") final FormDataBodyPart udfConfig,
                                   @FormDataParam("builtin") final boolean builtin,
                                   @Context SecurityContext securityContext); @Timed @GET @Produces({"application/java-archive", "application/json"}) @Path("/udfs/download/{udfId}") Response downloadUdf(@PathParam("udfId") Long udfId, @Context SecurityContext securityContext);  }