title: SCDF Aggregator Emits only the last record
body: <p>I am currently experimenting for one of the usecase we have to aggregate multiple input meesages into one before it gets written to a file system. Towards this I am trying to use the aggregator module available here
  <a href="https://repo.spring.io/release/org/springframework/cloud/stream/app/aggregator-processor-kafka/2.1.1.RELEASE/" rel="nofollow noreferrer">https://repo.spring.io/release/org/springframework/cloud/stream/app/aggregator-processor-kafka/2.1.1.RELEASE/</a></p>
  
  <p>My Environment is local SCDF/Kafka running on docker Dataflow 2.4.1.Release and Skipper 2.3.1.Release</p>
  
  <p>the stream definition is simple as below</p>
  
  <p>:inputTopic > aggregator > :outputTopic</p>
  
  <p>In the aggregator properties am providing these property values. </p>
  
  <pre><code>release: size() == 5
  group-timeout: 5000
  correlation: headers.headerName
  aggregation: T(org.springframework.util.StringUtils).collectionToDelimitedString(#this.![new String(payload)],'')
  message-store-type: simple
  </code></pre>
  
  <p>There are two issues I have </p>
  
  <ol>
  <li>Though i have given a release size of 5, and if there are say 10
  messages, i do not see my output topic to receive two messages</li>
  <li>The output topic contains only the last message of an aggregated header and not concatenated values of this list.</li>
  </ol>
  
  <p>Please suggest if there are any additional configurations that would help achieve an aggregated output. </p>
  
  <p>One additional observation is that if i have a stream as </p>
  
  <pre><code>http | aggregator &gt; :topic
  </code></pre>
  
  <p>The same aggregator config works fine and as desired. But the way we have it, the data is put into a topic via a processor using KStreams and not sure if there is a way in which the headers are generated and read using these default boot apps to a custom processor. </p>
  
  <p>One additional observation during debug of aggregator is that, the correlationKey is obtained or compared via hashCode and not as string, not sure if its a bug or a feature. but this is what is causing the disruption of expected behavior.</p>