Hi 大森林, if you look in the full logs you'll see
3989 [main] INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.apache.flink.test.checkpointing.UserActionLogPOJO does not contain a getter for field itemId 3999 [main] INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - class org.apache.flink.test.checkpointing.UserActionLogPOJO does not contain a setter for field itemId 3999 [main] INFO org.apache.flink.api.java.typeutils.TypeExtractor [] - Class class org.apache.flink.test.checkpointing.UserActionLogPOJO cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink documentation on "Data Types & Serialization" for details of the effect on performance. Exception in thread "main" org.apache.flink.api.common.typeutils.CompositeType$InvalidFieldReferenceException: Cannot reference field by field expression on GenericType<org.apache.flink.test.checkpointing.UserActionLogPOJO>Field expressions are only supported on POJO types, tuples, and case classes. (See the Flink documentation on what is considered a POJO.) at org.apache.flink.streaming.util.typeutils.FieldAccessorFactory.getAccessor(FieldAccessorFactory.java:193) at org.apache.flink.streaming.api.functions.aggregation.ComparableAggregator.<init>(ComparableAggregator.java:67) at org.apache.flink.streaming.api.datastream.KeyedStream.max(KeyedStream.java:809) at org.apache.flink.test.checkpointing.Aggregate.main(UnalignedCheckpointITCase.java:701) The issue is that in your POJO your setter/getter do not match the field names. It's easiest to let your IDE generate them for you. For example, if you keep the current field names, you need to add public String getItemId() { return itemId; } public void setItemId(String itemId) { this.itemId = itemId; } public void setPrice(int price) { this.price = price; } public void setUserId(String userId) { this.userId = userId; } public int getPrice() { return price; } public String getUserId() { return userId; } As you can see, none of your getters/setters is according to Java Beans specification and you need to add them all. On Fri, Oct 2, 2020 at 4:39 PM 大森林 <appleyu...@foxmail.com> wrote: > > I want to do an experiment with the operator "aggregate" > > My code is: > > Aggregate.java > https://paste.ubuntu.com/p/vvMKqZXt3r/ > > UserActionLogPOJO.java > https://paste.ubuntu.com/p/rfszzKbxDC/ > > > The error I got is: > > Exception in thread "main" > org.apache.flink.api.common.typeutils.CompositeType$InvalidFieldReferenceException: > Cannot reference field by field expression on > GenericType<UserActionLogPOJO>Field expressions are only supported on POJO > types, tuples, and case classes. (See the Flink documentation on what is > considered a POJO.) > at > org.apache.flink.streaming.util.typeutils.FieldAccessorFactory.getAccessor(FieldAccessorFactory.java:193) > at > org.apache.flink.streaming.api.functions.aggregation.ComparableAggregator.<init>(ComparableAggregator.java:67) > at > org.apache.flink.streaming.api.datastream.KeyedStream.max(KeyedStream.java:836) > at Aggregate.main(Aggregate.java:52) > > Process finished with exit code 1 > > *Could you tell me where I am wrong with UserActionLogPOJO.java?* > > *Thanks for your help* > > > -- Arvid Heise | Senior Java Developer <https://www.ververica.com/> Follow us @VervericaData -- Join Flink Forward <https://flink-forward.org/> - The Apache Flink Conference Stream Processing | Event Driven | Real Time -- Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany -- Ververica GmbH Registered at Amtsgericht Charlottenburg: HRB 158244 B Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng