GoogleHadoopFileSystemConfiguration { @VisibleForTesting static GoogleCloudStorageOptions.Builder getGcsOptionsBuilder(Configuration config) { String projectId = GCS_PROJECT_ID.get(config, config::get); return GoogleCloudStorageOptions.builder() .setStorageRootUrl(GCS_ROOT_URL.get(config, config::get)) .setStorageServicePath(GCS_SERVICE_PATH.get(config, config::get)) .setAutoRepairImplicitDirectoriesEnabled( GCS_REPAIR_IMPLICIT_DIRECTORIES_ENABLE.get(config, config::getBoolean)) .setInferImplicitDirectoriesEnabled( GCS_INFER_IMPLICIT_DIRECTORIES_ENABLE.get(config, config::getBoolean)) .setCopyWithRewriteEnabled(GCS_COPY_WITH_REWRITE_ENABLE.get(config, config::getBoolean)) .setMaxBytesRewrittenPerCall(GCS_REWRITE_MAX_BYTES_PER_CALL.get(config, config::getLong)) .setCopyMaxRequestsPerBatch(GCS_COPY_MAX_REQUESTS_PER_BATCH.get(config, config::getLong)) .setCopyBatchThreads(GCS_COPY_BATCH_THREADS.get(config, config::getInt)) .setOverwriteGenerationMismatchIgnored( GCS_OVERWRITE_GENERATION_MISMATCH_IGNORE.get(config, config::getBoolean)) .setTransportType( HTTP_TRANSPORT_SUFFIX.withPrefixes(CONFIG_KEY_PREFIXES).get(config, config::getEnum)) .setProxyAddress( PROXY_ADDRESS_SUFFIX.withPrefixes(CONFIG_KEY_PREFIXES).get(config, config::get)) .setProxyUsername( RedactedString.create( PROXY_USERNAME_SUFFIX.withPrefixes(CONFIG_KEY_PREFIXES).getPassword(config))) .setProxyPassword( RedactedString.create( PROXY_PASSWORD_SUFFIX.withPrefixes(CONFIG_KEY_PREFIXES).getPassword(config))) .setProjectId(projectId) .setMaxListItemsPerCall(GCS_MAX_LIST_ITEMS_PER_CALL.get(config, config::getLong)) .setMaxRequestsPerBatch(GCS_MAX_REQUESTS_PER_BATCH.get(config, config::getLong)) .setBatchThreads(GCS_BATCH_THREADS.get(config, config::getInt)) .setMaxHttpRequestRetries(GCS_HTTP_MAX_RETRY.get(config, config::getInt)) .setHttpRequestConnectTimeout(GCS_HTTP_CONNECT_TIMEOUT.get(config, config::getInt)) .setHttpRequestReadTimeout(GCS_HTTP_READ_TIMEOUT.get(config, config::getInt)) .setAppName(getApplicationName(config)) .setMaxWaitMillisForEmptyObjectCreation( GCS_MAX_WAIT_MILLIS_EMPTY_OBJECT_CREATE.get(config, config::getInt)) .setReadChannelOptions(getReadChannelOptions(config)) .setWriteChannelOptions(getWriteChannelOptions(config)) .setRequesterPaysOptions(getRequesterPaysOptions(config, projectId)) .setCooperativeLockingOptions(getCooperativeLockingOptions(config)) .setHttpRequestHeaders(GCS_HTTP_HEADERS.getPropsWithPrefix(config)) .setEncryptionAlgorithm(GCS_ENCRYPTION_ALGORITHM.get(config, config::get)) .setEncryptionKey(RedactedString.create(GCS_ENCRYPTION_KEY.getPassword(config))) .setEncryptionKeyHash(RedactedString.create(GCS_ENCRYPTION_KEY_HASH.getPassword(config))) .setGrpcEnabled(GCS_GRPC_ENABLE.get(config, config::getBoolean)) .setAuthorizationHandlerImplClass( GCS_AUTHORIZATION_HANDLER_IMPL.get( config, (k, d) -> config.getClass(k, d, AuthorizationHandler.class))) .setAuthorizationHandlerProperties( GCS_AUTHORIZATION_HANDLER_PROPERTIES_PREFIX.getPropsWithPrefix(config)); }   static final String GCS_CONFIG_PREFIX; static final List<String> CONFIG_KEY_PREFIXES; static final HadoopConfigurationProperty<String> GCS_ROOT_URL; static final HadoopConfigurationProperty<String> GCS_SERVICE_PATH; static final HadoopConfigurationProperty<String> PERMISSIONS_TO_REPORT; static final HadoopConfigurationProperty<Long> BLOCK_SIZE; static final HadoopConfigurationProperty<String> DELEGATION_TOKEN_BINDING_CLASS; static final HadoopConfigurationProperty<String> GCS_PROJECT_ID; static final HadoopConfigurationProperty<String> GCS_WORKING_DIRECTORY; static final HadoopConfigurationProperty<Boolean> GCE_BUCKET_DELETE_ENABLE; static final HadoopConfigurationProperty<RequesterPaysMode> GCS_REQUESTER_PAYS_MODE; static final HadoopConfigurationProperty<String> GCS_REQUESTER_PAYS_PROJECT_ID; static final HadoopConfigurationProperty<Collection<String>> GCS_REQUESTER_PAYS_BUCKETS; static final HadoopConfigurationProperty<GcsFileChecksumType> GCS_FILE_CHECKSUM_TYPE; static final HadoopConfigurationProperty<Boolean> GCS_PERFORMANCE_CACHE_ENABLE; static final HadoopConfigurationProperty<Long> GCS_PERFORMANCE_CACHE_MAX_ENTRY_AGE_MILLIS; static final HadoopConfigurationProperty<Boolean> GCS_STATUS_PARALLEL_ENABLE; static final HadoopConfigurationProperty<Boolean> GCS_LAZY_INITIALIZATION_ENABLE; static final HadoopConfigurationProperty<Boolean> GCS_REPAIR_IMPLICIT_DIRECTORIES_ENABLE; static final HadoopConfigurationProperty<Boolean> GCS_INFER_IMPLICIT_DIRECTORIES_ENABLE; static final HadoopConfigurationProperty<Boolean> GCS_FLAT_GLOB_ENABLE; static final HadoopConfigurationProperty<Boolean> GCS_CONCURRENT_GLOB_ENABLE; static final HadoopConfigurationProperty<String> GCS_MARKER_FILE_PATTERN; static final HadoopConfigurationProperty<Long> GCS_MAX_REQUESTS_PER_BATCH; static final HadoopConfigurationProperty<Integer> GCS_BATCH_THREADS; static final HadoopConfigurationProperty<Long> GCS_COPY_MAX_REQUESTS_PER_BATCH; static final HadoopConfigurationProperty<Integer> GCS_COPY_BATCH_THREADS; static final HadoopConfigurationProperty<Boolean> GCS_COPY_WITH_REWRITE_ENABLE; static final HadoopConfigurationProperty<Long> GCS_REWRITE_MAX_BYTES_PER_CALL; static final HadoopConfigurationProperty<Boolean> GCS_OVERWRITE_GENERATION_MISMATCH_IGNORE; static final HadoopConfigurationProperty<Long> GCS_MAX_LIST_ITEMS_PER_CALL; static final HadoopConfigurationProperty<Integer> GCS_HTTP_MAX_RETRY; static final HadoopConfigurationProperty<Integer> GCS_HTTP_CONNECT_TIMEOUT; static final HadoopConfigurationProperty<Integer> GCS_HTTP_READ_TIMEOUT; static final HadoopConfigurationProperty<String> GCS_APPLICATION_NAME_SUFFIX; static final HadoopConfigurationProperty<Integer> GCS_MAX_WAIT_MILLIS_EMPTY_OBJECT_CREATE; static final HadoopConfigurationProperty<OutputStreamType> GCS_OUTPUT_STREAM_TYPE; static final HadoopConfigurationProperty<Integer> GCS_OUTPUT_STREAM_BUFFER_SIZE; static final HadoopConfigurationProperty<Integer> GCS_OUTPUT_STREAM_PIPE_BUFFER_SIZE; static final HadoopConfigurationProperty<PipeType> GCS_OUTPUT_STREAM_PIPE_TYPE; static final HadoopConfigurationProperty<Integer> GCS_OUTPUT_STREAM_UPLOAD_CHUNK_SIZE; static final HadoopConfigurationProperty<Integer> GCS_OUTPUT_STREAM_UPLOAD_CACHE_SIZE; static final HadoopConfigurationProperty<Boolean> GCS_OUTPUT_STREAM_DIRECT_UPLOAD_ENABLE; static final HadoopConfigurationProperty<Integer> GCS_OUTPUT_STREAM_SYNC_MIN_INTERVAL_MS; static final HadoopConfigurationProperty<Integer> GCS_INPUT_STREAM_BUFFER_SIZE; static final HadoopConfigurationProperty<Boolean> GCS_INPUT_STREAM_FAST_FAIL_ON_NOT_FOUND_ENABLE; static final HadoopConfigurationProperty<Boolean> GCS_INPUT_STREAM_SUPPORT_GZIP_ENCODING_ENABLE; static final HadoopConfigurationProperty<Long> GCS_INPUT_STREAM_INPLACE_SEEK_LIMIT; static final HadoopConfigurationProperty<Fadvise> GCS_INPUT_STREAM_FADVISE; static final HadoopConfigurationProperty<Integer> GCS_INPUT_STREAM_MIN_RANGE_REQUEST_SIZE; static final HadoopConfigurationProperty<Boolean> GCS_GRPC_ENABLE; static final HadoopConfigurationProperty<Boolean> GCS_GRPC_CHECKSUMS_ENABLE; static final HadoopConfigurationProperty<String> GCS_GRPC_SERVER_ADDRESS; static final HadoopConfigurationProperty<String> GCS_CONFIG_OVERRIDE_FILE; static final HadoopConfigurationProperty<Boolean> GCS_COOPERATIVE_LOCKING_ENABLE; static final HadoopConfigurationProperty<Long> GCS_COOPERATIVE_LOCKING_EXPIRATION_TIMEOUT_MS; static final HadoopConfigurationProperty<Integer> GCS_COOPERATIVE_LOCKING_MAX_CONCURRENT_OPERATIONS; static final HadoopConfigurationProperty<Map<String, String>> GCS_HTTP_HEADERS; static final HadoopConfigurationProperty<String> GCS_ENCRYPTION_ALGORITHM; static final HadoopConfigurationProperty<RedactedString> GCS_ENCRYPTION_KEY; static final HadoopConfigurationProperty<RedactedString> GCS_ENCRYPTION_KEY_HASH; static final HadoopConfigurationProperty<Class<? extends AuthorizationHandler>> GCS_AUTHORIZATION_HANDLER_IMPL; static final HadoopConfigurationProperty<Map<String, String>> GCS_AUTHORIZATION_HANDLER_PROPERTIES_PREFIX; }
@Test public void customPropertiesValues() { Configuration config = new Configuration(); config.set(GCS_ROOT_URL.getKey(), "https: config.set(GCS_SERVICE_PATH.getKey(), "storage/dev_v1/"); GoogleCloudStorageOptions options = GoogleHadoopFileSystemConfiguration.getGcsOptionsBuilder(config).build(); assertThat(options.getStorageRootUrl()).isEqualTo("https: assertThat(options.getStorageServicePath()).isEqualTo("storage/dev_v1/"); }
@Test public void testGetAuthorizationHandler() { Configuration config = new Configuration(); config.setClass( GCS_AUTHORIZATION_HANDLER_IMPL.getKey(), FakeAuthorizationHandler.class, AuthorizationHandler.class); GoogleCloudStorageOptions options = GoogleHadoopFileSystemConfiguration.getGcsOptionsBuilder(config).build(); Class<? extends AuthorizationHandler> handler = options.getAuthorizationHandlerImplClass(); assertThat(handler).isAssignableTo(AuthorizationHandler.class); assertThat(handler).isEqualTo(FakeAuthorizationHandler.class); }
@Test public void testAuthorizationHandlerClassNotFound() { Configuration config = new Configuration(); config.set(GCS_AUTHORIZATION_HANDLER_IMPL.getKey(), "test.class.not.exist"); assertThrows( RuntimeException.class, () -> GoogleHadoopFileSystemConfiguration.getGcsOptionsBuilder(config).build()); }
@Test public void testProxyProperties_throwsExceptionWhenMissingProxyAddress() { HadoopConfigurationProperty<String> gcsProxyUsername = new HadoopConfigurationProperty<>( GCS_CONFIG_PREFIX + PROXY_USERNAME_SUFFIX.getKey(), "proxy-user"); HadoopConfigurationProperty<String> gcsProxyPassword = new HadoopConfigurationProperty<>( GCS_CONFIG_PREFIX + PROXY_PASSWORD_SUFFIX.getKey(), "proxy-pass"); Configuration config = new Configuration(); config.set(gcsProxyUsername.getKey(), gcsProxyUsername.getDefault()); config.set(gcsProxyPassword.getKey(), gcsProxyPassword.getDefault()); GoogleCloudStorageOptions.Builder optionsBuilder = GoogleHadoopFileSystemConfiguration.getGcsOptionsBuilder(config); assertThrows(IllegalArgumentException.class, optionsBuilder::build); }
@Test public void testDeprecatedKeys_throwsExceptionWhenDeprecatedKeyIsUsed() { HadoopConfigurationProperty<String> gcsProxyAddress = new HadoopConfigurationProperty<>( GCS_CONFIG_PREFIX + PROXY_ADDRESS_SUFFIX.getKey(), "proxy-address", "fs.gs.proxy.deprecated.address"); HadoopConfigurationProperty<Integer> gcsProxyUsername = new HadoopConfigurationProperty<>( GCS_CONFIG_PREFIX + PROXY_USERNAME_SUFFIX.getKey(), 1234, "fs.gs.proxy.deprecated.user"); HadoopConfigurationProperty<String> gcsProxyPassword = new HadoopConfigurationProperty<>( GCS_CONFIG_PREFIX + PROXY_PASSWORD_SUFFIX.getKey(), "proxy-pass", "fs.gs.proxy.deprecated.pass"); Configuration config = new Configuration(); config.set(gcsProxyAddress.getKey(), gcsProxyAddress.getDefault()); config.setInt(gcsProxyUsername.getKey(), gcsProxyUsername.getDefault()); config.set("fs.gs.proxy.deprecated.pass", gcsProxyPassword.getDefault()); String userPass = gcsProxyPassword.getPassword(config); assertThat(userPass).isEqualTo("proxy-pass"); GoogleCloudStorageOptions.Builder optionsBuilder = GoogleHadoopFileSystemConfiguration.getGcsOptionsBuilder(config); assertThrows(IllegalArgumentException.class, optionsBuilder::build); }