Page History

...

header.config

The format for response.http.headers will header.config 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. 

...

It is optional. it is a comma separated values of included path specs applied to headersheader.config. See path spec rules section.

...

It is optional. it is a comma separated values of excluded path specs applied to headersheader.config. See path spec rules section.

...

It is optional. it is a comma separated values of included mime types applied to headersheader.config.

Example:

included.mime.types=application/json

...

It is optional. it is a comma separated values of excluded mime types applied to headersheader.config.

Example:

excluded.mime.types=application/xml

...

It is optional. it is a comma separated values of included http methods applied to headersheader.config.

Example:

included.http.methods=POST,PUT

...

It is optional. it is a comma separated values of excluded http methods applied to headersheader.config.

Example:

excluded.http.methods=GET

...

We define a prefix response.http.headers.<name>. to allow multiple configurations, so that different paths have different headers. The "<name>" will be the name of header and that will be used to uniquely identify the header config which could only apply to different Servlet path or HTTP methods. 

Example:

response.http.headers.header1.header.config=set X-Frame-Options: DENY, "add Cache-Control: no-cache, no-store, must-revalidate", setDate Expires: 31540000000, addDate Last-Modified: 0
response.http.headers.header1.header.included.paths=^/test/0$
response.http.headers.header1.excluded.paths=^/test1/0$     
response.http.headers.header1.included.mime.types=application/jsonheaderjson
response.http.headers.header1.excluded.mime.types=application/xml
response.http.headers.header1.included.http.methods=POST,PUT
response.http.headers.header1.excluded.http.methods=GET
response.http.headers.header2.header.config=set X-Frame-Options: DENY, "add Cache-Control: no-cache, no-store, must-revalidate", setDate Expires: 31540000000, addDate Last-Modified: 0
response.http.headers.header2.header.included.paths=^/test2/0$
response.http.headers.header2.excluded.paths=^/test3/0$     
response.http.headers.header2.included.mime.types=application/json
response.http.headers.header2.excluded.mime.types=application/xml
response.http.headers.header2.included.http.methods=POST,PUT
response.http.headers.header2.excluded.http.methods=GET

Implementation

Implementation will use the Jetty HeaderFilter class. We need to update org.apache.kafka.connect.runtime.rest.RestServer class. During initialization the Connect REST server will read all header configurations from the property with prefix response.http.headers., then create a list of FilterHolder with HeaderFilter class and add the list of filter holders to the Servlet context handler based on the name of the  filter. Implementation is similar to how we handle the header access.control.allow.origin in the Connect REST server.

Pseudocode

ServletContextHandlerpublic context = new ServletContextHandlervoid configureHttpResponsHeaderFilters(ServletContextHandler.SESSIONS);
//Transfer WorkConfig to map which map filter name to a map which map name of init parameter to value of init parameter for the filter.   
 context) {
  Map<String, Map<String, String>> headerFilterConfigsheadersConfigs = extractHeaderFilterConfigextractHttpResponseHeaderConfig(workerConfig);
  FilterHolder headersFilterHolderheaderFilterHolder = null;
  for (Map.Entry<String, Map<String, String>> oneFilterentry : headerFilterConfigsheadersConfigs.getEntrySetentrySet()) {
    headersFilterHolderheaderFilterHolder = new FilterHolder(HeaderFilter.class);
  context.addFilter(headersFilterHolder, "/*", EnumSetheaderFilterHolder.ofsetName(DispatcherTypeentry.REQUESTgetKey()); 
  headersFilterHolder.setName(oneFilter.getName());
  Map<String, String> oneHeaderConfig = oneFilterentry.getValue());
    for (Map.Entry<String, String> oneHeader : oneHeaderConfig.forEach((k,v) ->entrySet()) {
      switch (oneHeader.getKey(k).toUpperCase()) {
        case "HEADER.CONFIG":
          headersFilterHolderheaderFilterHolder.setInitParameter("headerConfig", voneHeader.getValue());
          break;
        case "INCLUDEEDINCLUDED.PATHS":
          headersFilterHolderheaderFilterHolder.setInitParameter("includedPaths", voneHeader.getValue());
          break;
        case "EXCLUDEEDEXCLUDED.PATHS":
          headersFilterHolderheaderFilterHolder.setInitParameter("excludedPathsexcludedPaths", voneHeader.getValue());
          break;
        case "INCLUDED.MIME.TYPES":
          headersFilterHolderheaderFilterHolder.setInitParameter("includedMimeTypes", voneHeader.getValue());
          break;
        case "ENCLUDED.MIME.TYPES":
          headersFilterHolderheaderFilterHolder.setInitParameter("excludedMimeTypes", voneHeader.getValue());
          break;
        case "INCLUDED.HTTP.METHODS":
          headersFilterHolderheaderFilterHolder.setInitParameter("includedHttpMeethodsincludedHttpMethods", voneHeader.getValue());
          break;
        case "ENCLUDED.HTTP.METHODS":
          headersFilterHolderheaderFilterHolder.setInitParameter("excludedHttpMeethodsexcludedHttpMethods", voneHeader.getValue());
          break;
        default:
      }
    }
    context.addFilter(headerFilterHolder, "/*", EnumSet.of(DispatcherType.REQUEST));
  }
}

References

https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers
https://www.eclipse.org/jetty/documentation/current/header-filter.html
https://www.eclipse.org/jetty/javadoc/9.4.24.v20191120/org/eclipse/jetty/servlets/HeaderFilter.html

...