class HttpRequestParser implements RequestParser<HttpHeaders> {

    @Override
    public RequestContext parseRequest(HttpHeaders request, RequestContext.RequestType requestType) {

        RequestMap params = new RequestMap(request.getRequestHeaders());

        if (LOG.isDebugEnabled()) {
            // Logging only keys to prevent sensitive data to be logged
            LOG.debug("Parsing request parameters: " + params.keySet());
        }

        // build new instance of RequestContext and fill it with parsed values
        RequestContext context = new RequestContext();

        // whether we are in a fragmenter, read_bridge, or write_bridge scenario
        context.setRequestType(requestType);

        // first of all, set profile and enrich parameters with information from specified profile
        String profile = params.removeUserProperty("PROFILE");
        context.setProfile(profile);
        addProfilePlugins(profile, params);

        // Ext table uses system property FORMAT for wire serialization format
        String wireFormat = params.removeProperty("FORMAT");
        context.setOutputFormat(OutputFormat.valueOf(wireFormat));

        // FDW uses user property FORMAT to indicate format of data
        String format = params.removeUserProperty("FORMAT");
        format = StringUtils.isNotBlank(format) ? format : context.inferFormatName();
        context.setFormat(format);

        context.setAccessor(params.removeUserProperty("ACCESSOR"));
        context.setAggType(EnumAggregationType.getAggregationType(params.removeOptionalProperty("AGG-TYPE")));

        /*
         * Some resources don't require a fragment, hence the list can be empty.
         */
        String fragmentStr = params.removeOptionalProperty("DATA-FRAGMENT");
        if (StringUtils.isNotBlank(fragmentStr)) {
            context.setDataFragment(Integer.parseInt(fragmentStr));
        }

        context.setDataSource(params.removeProperty("DATA-DIR"));

        String filterString = params.removeOptionalProperty("FILTER");
        String hasFilter = params.removeProperty("HAS-FILTER");
        if (filterString != null) {
            context.setFilterString(filterString);
        } else if ("1".equals(hasFilter)) {
            LOG.info("Original query has filter, but it was not propagated to PXF");
        }

        context.setFragmenter(params.removeUserProperty("FRAGMENTER"));

        String fragmentIndexStr = params.removeOptionalProperty("FRAGMENT-INDEX");
        if (StringUtils.isNotBlank(fragmentIndexStr)) {
            context.setFragmentIndex(Integer.parseInt(fragmentIndexStr));
        }

        String encodedFragmentMetadata = params.removeOptionalProperty("FRAGMENT-METADATA");
        context.setFragmentMetadata(Utilities.parseBase64(encodedFragmentMetadata, "Fragment metadata information"));
        context.setHost(params.removeProperty("URL-HOST"));
        context.setMetadata(params.removeUserProperty("METADATA"));
        context.setPort(params.removeIntProperty("URL-PORT"));
        context.setProfileScheme(params.removeUserProperty(PROFILE_SCHEME));
        context.setProtocol(params.removeUserProperty("PROTOCOL"));
        context.setRemoteLogin(params.removeOptionalProperty("REMOTE-USER"));
        context.setRemoteSecret(params.removeOptionalProperty("REMOTE-PASS"));
        context.setResolver(params.removeUserProperty("RESOLVER"));
        context.setSegmentId(params.removeIntProperty("SEGMENT-ID"));
        context.setServerName(params.removeUserProperty("SERVER"));

        // An optional CONFIG value specifies the name of the server
        // configuration directory, if not provided the config is the server name
        String config = params.removeUserProperty("CONFIG");
        context.setConfig(StringUtils.isNotBlank(config) ? config : context.getServerName());

        String maxFrags = params.removeUserProperty("STATS-MAX-FRAGMENTS");
        if (!StringUtils.isBlank(maxFrags)) {
            context.setStatsMaxFragments(Integer.parseInt(maxFrags));
        }

        String sampleRatioStr = params.removeUserProperty("STATS-SAMPLE-RATIO");
        if (!StringUtils.isBlank(sampleRatioStr)) {
            context.setStatsSampleRatio(Float.parseFloat(sampleRatioStr));
        }

        context.setTotalSegments(params.removeIntProperty("SEGMENT-COUNT"));
        context.setTransactionId(params.removeProperty("XID"));

        // parse tuple description
        parseTupleDescription(params, context);

        if (context.getOutputFormat() == OutputFormat.TEXT) {
            // parse CSV format information
            parseGreenplumCSV(params, context);

            // Only single column tables support 'OFF' delimiter
            if (context.getTupleDescription().size() != 1 && context.getGreenplumCSV().getDelimiter() == null) {
                throw new IllegalArgumentException(String.format("using no delimiter is only possible for a single column table. %d columns found", context.getTupleDescription().size()));
            }
        }

        context.setUser(params.removeProperty("USER"));

        String encodedFragmentUserData = params.removeOptionalProperty("FRAGMENT-USER-DATA");
        context.setUserData(Utilities.parseBase64(encodedFragmentUserData, "Fragment user data"));

        // Store alignment for global use as a system property
        System.setProperty("greenplum.alignment", params.removeProperty("ALIGNMENT"));

        Map<String, String> optionMappings = null;

        // prepare addition configuration properties if profile was specified
        if (StringUtils.isNotBlank(profile)) {
            optionMappings = pluginConf.getOptionMappings(profile);
        }

        // use empty map for convenience instead of null
        if (optionMappings == null) {
            optionMappings = Collections.emptyMap();
        }

        Map<String, String> additionalConfigProps = new HashMap<>();

        // Iterate over the remaining properties
        // we clone the keyset to prevent concurrent modification exceptions
        List<String> paramNames = new ArrayList<>(params.keySet());
        for (String param : paramNames) {
            if (StringUtils.startsWithIgnoreCase(param, RequestMap.USER_PROP_PREFIX)) {
                // Add all left-over user properties as options
                String optionName = param.toLowerCase().replace(RequestMap.USER_PROP_PREFIX_LOWERCASE, "");
                String optionValue = params.removeUserProperty(optionName);
                context.addOption(optionName, optionValue);
                LOG.debug("Added option {} to request context", optionName);

                // lookup if the option should also be applied as a config property
                String propertyName = optionMappings.get(optionName);
                if (StringUtils.isNotBlank(propertyName)) {
                    // if option has been provided by the user in the request, set the value
                    // of the corresponding configuration property
                    if (optionValue != null) {
                        additionalConfigProps.put(propertyName, optionValue);
                        // do not log property value as it might contain sensitive information
                        LOG.debug("Added extra config property {} from option {}", propertyName, optionName);
                    }
                }
            } else if (StringUtils.startsWithIgnoreCase(param, RequestMap.PROP_PREFIX)) {
                // log debug for all left-over system properties
                LOG.debug("Unused property {}", param);
            }
        }

        context.setAdditionalConfigProps(additionalConfigProps);
        context.setPluginConf(pluginConf);

        // Call the protocol handler for any protocol-specific logic handling
        if (StringUtils.isNotBlank(profile)) {
            String handlerClassName = pluginConf.getHandler(profile);
            if (StringUtils.isNotBlank(handlerClassName)) {
                Class<?> clazz;
                try {
                    clazz = Class.forName(handlerClassName);
                    ProtocolHandler handler = (ProtocolHandler) clazz.getDeclaredConstructor().newInstance();
                    context.setFragmenter(handler.getFragmenterClassName(context));
                    context.setAccessor(handler.getAccessorClassName(context));
                    context.setResolver(handler.getResolverClassName(context));
                } catch (ClassNotFoundException | NoSuchMethodException | InvocationTargetException |
                        InstantiationException | IllegalAccessException e) {
                    throw new RuntimeException(String.format("Error when invoking handlerClass '%s' : %s", handlerClassName, e), e);
                }
            }
        }

        // validate that the result has all required fields, and values are in valid ranges
        context.validate();

        return context;
    }

    public  HttpRequestParser();
      HttpRequestParser(PluginConf pluginConf);

}