Swap http for sdk client

[xShinnRyuu]

Issue #, if available:

Description of changes:
Switched Bulk Loading request to use AWS SDK NeptunedataClient instead of http requests.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[andreachild]

Nit: just a thought, would it be cleaner to just to have a reference to the BulkLoadConfig instead of copying many values as fields here? For example:

private final Region region;
private final BulkLoadConfig config;

public NeptuneBulkLoader(BulkLoadConfig bulkLoadConfig) {
    this.region = extractRegionFromEndpoint(bulkLoadConfig.getNeptuneEndpoint());
    this.config = bulkLoadConfig;

And then usage of the values would just need to call the getters from the BulkLoadConfig instead (sanitizing the value is necessary).

[xShinnRyuu]

Definitely cleaner. I will update it

[andreachild]

I noticed the NEPTUNE_PORT is hardcoded to 8182 which is the default. It is possible for the port to be modified as neptune config OR the user using this tool could be using ssh tunnel with a different port.

[xShinnRyuu]

thanks for bringing that up. I have added it as a configurable parameter in the config and as a cli command

[andreachild]

You should be able to handle the cleanup with a finally block.

Suggested change

	.whenComplete((result, throwable) -> closeStreams(streamExecutor, pipedOut, pipedIn));
	} catch (Exception e) {
	} finally {

[xShinnRyuu]

Refactored how the method does the error catching. It now just throws it up to the caller method and has all of the logging and exception checking done in there.

Removed the catch block and added a finally to close off the thread & streams

[andreachild]

Curious, did try with resources not work? Something like:

ExecutorService streamExecutor = Executors.newSingleThreadExecutor();
        try (PipedOutputStream pipedOut = new PipedOutputStream();
             PipedInputStream pipedIn = new PipedInputStream(pipedOut)) {
            CompletableFuture<Void> compressionFuture = startCompressionTask(localFile, pipedOut);
            UploadRequest uploadRequest = createUploadRequest(s3Key, pipedIn, streamExecutor);

            System.err.println("Initiating Transfer Manager upload...");
            Upload upload = transferManager.upload(uploadRequest);

            return CompletableFuture.allOf(upload.completionFuture(), compressionFuture)
                    .thenApply(ignored -> handleUploadSuccess(localFile, upload))
                    .exceptionally(throwable -> handleUploadFailure(localFilePath, throwable));
        } finally {
            streamExecutor.shutdown();
        }

[xShinnRyuu]

ExecutorService does not have autocloseable implemented in Java 17 (only after Java 19). Since i had to pull this thread creation out, it was doing some funny interactions with the pipes being declared in the autocloseable try block. The pipes would either close before every thing was finish or just would not connect.

This led me to moving both of the input and output pipes out of the try block to fix the sync issue.