Developing Applications On Managed Service For Apache Flink

// Recommended: Flink POJO for optimal performance with schema evolution
public class OptimizedEvent {
    // All fields must be public or have public getters/setters
    public String eventId;
    public long timestamp;
    public String userId;
    public EventType type;
    
    // Required: public no-argument constructor
    public OptimizedEvent() {}
    
    public OptimizedEvent(String eventId, long timestamp, String userId, EventType type) {
        this.eventId = eventId;
        this.timestamp = timestamp;
        this.userId = userId;
        this.type = type;
    }
}

// Enum types work well with POJO serialization
public enum EventType {
    USER_ACTION, SYSTEM_EVENT, ERROR_EVENT
}

// Use when performance is critical and schema evolution is not needed
DataStream<Tuple4<String, Long, String, Integer>> events = source
    .map(event -> Tuple4.of(event.getId(), event.getTimestamp(), 
                           event.getUserId(), event.getCount()));

// Access fields by position (f0, f1, f2, f3)
events.keyBy(tuple -> tuple.f2) // Key by userId (f2)
      .process(new TupleProcessor());

// Use Avro when integrating with external systems or when advanced schema evolution is needed
public class AvroEventProcessor extends ProcessFunction<SpecificRecordBase, ProcessedEvent> {
    
    @Override
    public void processElement(SpecificRecordBase avroEvent, Context ctx, 
                              Collector<ProcessedEvent> out) {
        if (avroEvent instanceof UserEvent) {
            UserEvent userEvent = (UserEvent) avroEvent;
            ProcessedEvent result = new ProcessedEvent();
            result.setUserId(userEvent.getUserId().toString());
            result.setTimestamp(userEvent.getTimestamp());
            out.collect(result);
        }
    }
}

// Configure Avro serialization
// Note: enableForceAvro() is available in Flink 1.20 but removed in 2.x.
// For Flink 2.2, use AvroTypeInfo explicitly in state descriptors instead.
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Flink 1.20 only:
// env.getConfig().enableForceAvro();

// Recommended: Convert Protobuf to POJO at the boundary, avoiding Kryo entirely
DataStream<MyEvent> events = protobufSource
    .map(proto -> new MyEvent(
        proto.getEventId(),
        proto.getTimestamp(),
        proto.getUserId()));
// MyEvent is a Flink POJO (public fields + no-arg constructor) — fast serialization, schema evolution, no Kryo

// Monitor for Kryo fallbacks in logs - these indicate performance issues
// Log message: "Class ... cannot be used as a POJO type because not all fields are valid POJO fields"

// To detect Kryo usage, disable it temporarily during development
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableGenericTypes(); // Throws exception if Kryo would be used
// This will fail with: "Generic types have been disabled in the ExecutionConfig"

// Last resort — register frequently used types to avoid class name serialization overhead IF you use Kryo
// In Flink 2.x, use env.getConfig() methods (env-level convenience methods are removed)
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.getConfig().registerKryoType(CustomEvent.class);
env.getConfig().registerKryoType(ProcessingResult.class);

env.getConfig().registerTypeWithKryoSerializer(
    ComplexObject.class, 
    CustomKryoSerializer.class
);

// For state objects, prioritize schema evolution support
public class StatefulProcessor extends KeyedProcessFunction<String, Event, Result> {
    
    // Use POJO or Avro for state that needs to evolve
    private transient ValueState<EventAggregate> aggregateState; // POJO - good performance + evolution
    
    // Use primitive types for simple state
    private transient ValueState<Long> counterState; // Primitive - fastest
    
    @Override
    public void open(OpenContext openContext) throws Exception {
        // POJO state descriptor - supports schema evolution
        ValueStateDescriptor<EventAggregate> aggregateDescriptor = 
            new ValueStateDescriptor<>("aggregate", EventAggregate.class);
        aggregateState = getRuntimeContext().getState(aggregateDescriptor);
        
        // Primitive state descriptor - fastest serialization
        ValueStateDescriptor<Long> counterDescriptor = 
            new ValueStateDescriptor<>("counter", Long.class);
        counterState = getRuntimeContext().getState(counterDescriptor);
    }
}

// AVOID: Default Java Serialization (extremely slow)
public class SlowEvent implements Serializable {
    // Java serialization is 10x+ slower than POJO serialization
}

// AVOID: Complex nested objects without proper POJO structure
public class BadEvent {
    private Map<String, Object> data; // Generic Object causes Kryo fallback
    private List<SomeInterface> items; // Interface types cause Kryo fallback
}

// AVOID: Missing no-argument constructor (causes Kryo fallback)
public class InvalidPOJO {
    public String field;
    
    // Missing no-arg constructor - will use Kryo instead of POJO serializer
    public InvalidPOJO(String field) {
        this.field = field;
    }
}

// AVOID: Private fields without getters/setters (causes Kryo fallback)
public class AlmostPOJO {
    private String secretField; // No getter/setter - not a valid POJO field
    public String publicField;   // This is fine
    
    public AlmostPOJO() {} // Constructor is correct
}