Composite Aggregation after key ignored

What is the bug?

when searching with composite aggregation the supplied afterKey is ignored and we always get back the same initial afterKey.
Therefore, iterating with composite aggregation until the afterKey is null, leads to an infinite loop

client version

we encountered this bug with client version 2.6.0, 2.70 and also after upgrading to the latest 2.8.1 client version

How can one reproduce the bug?

search with composite aggregation in a loop until the afterKey is null.
Since the afterKey is ignored, it will always search from the beginning and repeatedly return the same afterKey

What is the expected behavior?

the afterKey should should be considered, so that after each search we get back the "next" afterKey

workaround

Unfortunately we switched back to the rest client :(

Code sample to reproduce infinite loop

List<Map<String, CompositeAggregationSource>> compositeAggSourceList = new ArrayList<>();

compositeAggSourceList.add(Collections.singletonMap("tenant_id", new CompositeAggregationSource.Builder()
				.terms(new TermsAggregation.Builder()
						.field("tenant_id")
						.build())
		.build()));
compositeAggSourceList.add(Collections.singletonMap("user_id", new CompositeAggregationSource.Builder()
		.terms(new TermsAggregation.Builder()
				.field("user_id")
				.build())
		.build()));
					
CompositeAggregate compositeAggregate = null;
do {
	// update the composite aggregation after key
	Map<String, String> afterKey;
	if(compositeAggregate !=null){
		afterKey = compositeAggregate.afterKey().entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
				e -> new String(e.getValue().toString())));
	} else {
		afterKey = null;
	}

	SearchRequest searchRequest = new SearchRequest.Builder()
			.index("example-index")
			.size(0)
			.aggregations("composite-agg", new Aggregation.Builder()
					.composite(c-> afterKey !=null ? c.sources(compositeAggSourceList).size(10).after(afterKey) : c.sources(compositeAggSourceList).size(10))
					.aggregations("incidents",
							a -> a.sum(t -> t.field("incidents")))
					.aggregations("activities",
							a -> a.sum(t -> t.field("activities")))
					.build())
			.build();

	SearchResponse<Map> searchResponse = openSearchClient.get().search(searchRequest, Map.class);

	compositeAggregate = searchResponse.aggregations().get("composite-agg").composite();
	
	// our internal logic...

} while (compositeAggregate.afterKey()!=null);

Start with the supplied Java reproduction and trace how SearchRequest builds the composite aggregation when an afterKey is present. Verify the request reaches the client correctly and reproduce the loop; done means successive searches honor the supplied afterKey and return the next page until afterKey is null.