I have the following situation in an Apache Flink project.

3 streams with different objects, like

**Person** -> String id, String firstName, String lastName (i.e. 101, John, Doe)

**PersonDetail** -> String id, String address, String city, String phoneNumber, long personId (i.e. 99, Stefansplatz 1, +43066012345678, 101)

**PersonAddDetail** -> String id, String AddDetailType, object AddDetailValue, long personId (i.e. 77, 1, Hansi or 78, 2, 1234 or 80, 3, true)

I'd like to aggregate (not sure if this is the right wording here) objects from these streams to a new object that I put to a new stream. The aggregation should be based on the Person id and as additional catch I need to filter out PersonAddDetail only with specific AddDetailType (let's say I'm only interested in objects with type 1 and 2).

The aggregated object should look somehow like 

**PersonReport** -> long id, String firstName, String lastName, String address, String city, String phoneNumber, ArrayList< PersonAddDetail > details

The question now is if this is possible at all and if yes how can I accomplish it. Every input welcome.