Best Practices

public class MSFStreamingApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // NO checkpoint configuration in code - managed by MSF service
        DataStream<Event> events = env
            .fromSource(createKinesisSource(), WatermarkStrategy.forMonotonousTimestamps(), "kinesis-source")
            .uid("kinesis-source-uid");
        
        DataStream<ProcessedEvent> processed = events
            .keyBy(Event::getKey)
            .process(new EventProcessor())
            .name("event-processor")
            .uid("event-processor-uid");
        
        processed.sinkTo(createS3Sink())
            .name("s3-sink")
            .uid("s3-sink-uid");
        
        env.execute("MSF Streaming Application");
    }
}

// AVOID: Single large operator doing everything
events.map(event -> {
    // Complex transformation logic
    // Multiple business rules
    // Data enrichment
    // Validation
    // Formatting
    return processedEvent;
}); // Hard to debug, scale, and maintain

// Application code should be KPU-agnostic
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// DO NOT set parallelism in application code for Managed Service for Apache Flink deployment
// Managed Service for Apache Flink manages parallelism through KPU configuration

// For local Docker development only:
if (isLocalDevelopment()) {
    int localParallelism = Math.max(1, Runtime.getRuntime().availableProcessors() - 1);
    env.setParallelism(localParallelism);
}

// Set operator-specific parallelism only when business logic or infrastructure requires it
// E.g. you may set parallelism on a Kafka source operator to be equal to number of partitions, if the overall app parallelism is higher than the number of partitions (lower parallelism for source, but enable high parallelism for processing operators with keyBy or similar operators that spread load)
dataStream
    .keyBy(Event::getPartitionKey)
    .process(new HeavyProcessor())
    // Only set if this operator specifically needs different parallelism
    .setParallelism(5);

public class RobustProcessor extends ProcessFunction<Event, ProcessedEvent> {
    public static final OutputTag<Event> DEAD_LETTER_TAG =
            new OutputTag<Event>("dead-letter") {};
    private transient Counter errorCounter;
    
    @Override
    public void open(OpenContext openContext) throws Exception {
        errorCounter = getRuntimeContext().getMetricGroup().counter("processing_errors");
    }
    
    @Override
    public void processElement(Event event, Context ctx, Collector<ProcessedEvent> out) {
        try {
            // Validate input
            if (!isValidEvent(event)) {
                LOG.warn("Invalid event received: {}", event);
                ctx.output(DEAD_LETTER_TAG, event);
                return;
            }
            
            ProcessedEvent result = processEvent(event);
            out.collect(result);
            
        } catch (TransientException e) {
            // Let Managed Service for Apache Flink handle transient errors through restart strategy
            LOG.warn("Transient error processing event {}, Managed Service for Apache Flink will retry", event.getId(), e);
            throw e; // Managed Service for Apache Flink restart strategy handles this
            
        } catch (Exception e) {
            // Handle permanent errors gracefully
            LOG.error("Permanent error processing event {}", event.getId(), e);
            errorCounter.inc();
            ctx.output(DEAD_LETTER_TAG, event);
            // Don't throw - continue processing other events
        }
    }
}

import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime;

public class FlinkStreamingJob {
    private static final String LOCAL_PROPS = "flink-application-properties-dev.json";
    
    private static boolean isLocal(StreamExecutionEnvironment env) {
        String runtime = System.getenv("RUNTIME_ENVIRONMENT");
        return env instanceof LocalStreamEnvironment || "local".equalsIgnoreCase(runtime);
    }

    private static Map<String, Properties> loadApplicationProperties(StreamExecutionEnvironment env) throws IOException {
        if (isLocal(env)) {
            InputStream input = FlinkStreamingJob.class.getClassLoader().getResourceAsStream(LOCAL_PROPS);
            if (input == null) throw new IOException("Unable to find " + LOCAL_PROPS);
            java.nio.file.Path tempFile = java.nio.file.Files.createTempFile("flink-app-props", ".json");
            java.nio.file.Files.copy(input, tempFile, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
            input.close();
            tempFile.toFile().deleteOnExit();
            return KinesisAnalyticsRuntime.getApplicationProperties(tempFile.toString());
        } else {
            return KinesisAnalyticsRuntime.getApplicationProperties();
        }
    }

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        if (isLocal(env)) {
            env.enableCheckpointing(10_000);
            env.setParallelism(3);
        }
        final Map<String, Properties> applicationProperties = loadApplicationProperties(env);
    }
}