...

There is no any changes on public interfaces. We just add a new configuration property "response.http.headers". The following section has detailed description on this new property.

Proposed Changes

Adding New Property

We will add a new property "response.http.headers" to in org.apache.kafka.connect.runtime.WorkerConfig class. It will allow REST server administrator to configure headers based on their security policies. We borrow and take advantage of Jetty HeaderFilter class and use same format of headerConfig init param. The format for response.http.headers will be "[[action] [header]:[header value],..." which is a list of [action] [header]:[value] separated by comma ",". So it is a CSV of actions to perform on headers with the following syntax:
[action] [header name]: [header value],
[action] can be one of "set, add, setDate, or addDate" which specify an action will perform on header. 

...

Implementation will use Jetty HeaderFilter class. We need update org.apache.kafka.connect.runtime.rest.RestServer class. During initializing process Connect REST server will read header configuration from the property response.http.headers, then create a FilterHoder with HeaderFilter class and add the filter holder to Servlet context handler. Implementation is similar as we handle header access.control.allow.origin in Connect REST server.

Pseudocode

ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS);
String responseHeaders = config.getString("response.http.headers");
FilterHolder headersFilterHolder = new FilterHolder(HeaderFilter.class);
headersFilterHolder.setName("headerConfig");
headersFilterHolder.setInitParameter("headerConfig", responseHeaders);
context.addFilter(headersFilterHolder, "/*", EnumSet.of(DispatcherType.REQUEST));

References

...