// SamzaConfig.java
import org.apache.samza.config.Config;
import org.apache.samza.config.MapConfig;
import org.apache.samza.serializers.JsonSerdeFactory;
import org.apache.samza.system.kafka.KafkaSystemFactory;
 
import java.util.HashMap;
import java.util.Map;
 
public class SamzaConfig {
    public static Config getConfig() {
        Map<String, String> configMap = new HashMap<>();
        configMap.put("job.name", "samza-flink-migration-example");
        configMap.put("job.factory.class", "org.apache.samza.job.yarn.YarnJobFactory");
        configMap.put("yarn.package.path", "/path/to/samza-job.tar.gz");
        configMap.put("task.inputs", "kafka.my-input-topic");
        configMap.put("task.output", "kafka.my-output-topic");
        configMap.put("serializers.registry.string.class", "org.apache.samza.serializers.StringSerdeFactory");
        configMap.put("serializers.registry.json.class", JsonSerdeFactory.class.getName());
        configMap.put("systems.kafka.samza.factory", KafkaSystemFactory.class.getName());
        configMap.put("systems.kafka.broker.list", "localhost:9092");
 
        return new MapConfig(configMap);
    }
}
 
// MySamzaTask.java
import org.apache.samza.application.StreamApplication;
import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
import org.apache.samza.config.Config;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.system.OutgoingMessageEnvelope;
import org.apache.samza.system.SystemStream;
import org.apache.samza.task.MessageCollector;
import org.apache.samza.task.TaskCoordinator;
import org.apache.samza.task.TaskContext;
import org.apache.samza.task.TaskInit;
import org.apache.samza.task.TaskRun;
import org.apache.samza.serializers.JsonSerde;
 
import java.util.HashMap;
import java.util.Map;
 
public class MySamzaTask implements StreamApplication, TaskInit, TaskRun {
    private JsonSerde<String> jsonSerde = new JsonSerde<>();
 
    @Override
    public void init(Config config, TaskContext context, TaskCoordinator coordinator) throws Exception {
        // Initialization logic if needed
    }
 
    @Override
    public void run() throws Exception {
        MessageCollector collector = getContext().getMessageCollector();
        SystemStream inputStream = getContext().getJobContext().getInputSystemStream("kafka", "my-input-topic");
 
        for (IncomingMessageEnvelope envelope : getContext().getPoll(inputStream, "MySamzaTask")) {
            String input = new String(envelope.getMessage());
            String output = processMessage(input);
            collector.send(new OutgoingMessageEnvelope(getContext().getOutputSystem("kafka"), "my-output-topic", jsonSerde.toBytes(output)));
        }
    }
 
    private String processMessage(String message) {
        // Simple processing logic: convert to uppercase
        return message.toUpperCase();
    }
 
    @Override
    public StreamApplicationDescriptor getDescriptor() {
        return new StreamApplicationDescriptor("MySamzaTask")
                .withConfig(SamzaConfig.getConfig())
                .withTaskClass(this.getClass());
    }
}

// FlinkConfig.java
import org.apache.flink.configuration.Configuration;
 
public class FlinkConfig {
    public static Configuration getConfig() {
        Configuration config = new Configuration();
        config.setString("execution.target", "streaming");
        config.setString("jobmanager.rpc.address", "localhost");
        config.setInteger("taskmanager.numberOfTaskSlots", 1);
        config.setString("pipeline.execution.mode", "STREAMING");
        return config;
    }
}
 
// MyFlinkJob.java
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
 
import java.util.Properties;
 
public class MyFlinkJob {
    public static void main(String[] args) throws Exception {
        // Set up the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        // Configure Kafka consumer
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-consumer-group");
 
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("my-input-topic", new SimpleStringSchema(), properties);
 
        // Add source
        DataStream<String> stream = env.addSource(consumer);
 
        // Process the stream
        DataStream<String> processedStream = stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return value.toUpperCase();
            }
        });
 
        // Configure Kafka producer
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>("my-output-topic", new SimpleStringSchema(), properties);
 
        // Add sink
        processedStream.addSink(producer);
 
        // Execute the Flink job
        env.execute("Flink Migration Example");
    }
}

# 編譯（假設使用Maven）
mvn clean package
 
# 提交到Flink集群（假設Flink在本地運行）
./bin/flink run -c com.example.MyFlinkJob target/your-jar-file.jar