拥抱AI-AI辅助测试与文档 —— 专业度的分水岭

package com.company.realtime.profile.process;import org.apache.flink.api.common.state.ValueStateDescriptor;import org.apache.flink.api.common.typeinfo.TypeInformation;import org.apache.flink.streaming.api.operators.KeyedProcessOperator;import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;import org.apache.flink.streaming.util.TestHarnessUtil;import org.junit.jupiter.api.BeforeEach;import org.junit.jupiter.api.Test;import java.util.concurrent.ConcurrentLinkedQueue;import static org.junit.jupiter.api.Assertions.*;class LongTermProfileFunctionTest {    private KeyedOneInputStreamOperatorTestHarness<String, UserAction, UserProfile> testHarness;    private LongTermProfileFunction processFunction;    @BeforeEach    void setUp() throws Exception {        processFunction = new LongTermProfileFunction();        testHarness = new KeyedOneInputStreamOperatorTestHarness<>(            new KeyedProcessOperator<>(processFunction),            UserAction::getUserId,            TypeInformation.of(String.class)        );        testHarness.open();    }    @Test    void testNormalProcessing() throws Exception {        // 准备测试数据        UserAction action1 = new UserAction("u1", ActionType.CLICK, "p1", "electronics", 0, 1000L);        UserAction action2 = new UserAction("u1", ActionType.PURCHASE, "p1", "electronics", 99.9, 2000L);        // 注入数据        testHarness.processElement(new StreamRecord<>(action1, 1000L));        testHarness.processElement(new StreamRecord<>(action2, 2000L));        // 模拟时间推进（触发定时器）        testHarness.setProcessingTime(61000L); // 1分钟后        // 收集输出        ConcurrentLinkedQueue<Object> output = testHarness.getOutput();        // 验证        assertFalse(output.isEmpty());        StreamRecord<UserProfile> record = (StreamRecord<UserProfile>) output.poll();        UserProfile profile = record.getValue();        assertEquals("u1", profile.getUserId());        assertEquals(1, profile.getClickCount());        assertEquals(99.9, profile.getPurchaseAmount());        assertEquals(UserProfile.WindowType.LONG, profile.getWindowType());    }    @Test    void testStateRecoveryAfterCheckpoint() throws Exception {        // 1. 处理第一条数据        UserAction action = new UserAction("u1", ActionType.CLICK, "p1", "electronics", 0, 1000L);        testHarness.processElement(new StreamRecord<>(action, 1000L));        // 2. 做checkpoint        testHarness.snapshot(1L, 1000L);        // 3. 关闭并恢复        testHarness.close();        testHarness.initializeState(1L);        testHarness.open();        // 4. 处理第二条数据        UserAction action2 = new UserAction("u1", ActionType.PURCHASE, "p1", "electronics", 99.9, 2000L);        testHarness.processElement(new StreamRecord<>(action2, 2000L));        // 5. 触发定时器        testHarness.setProcessingTime(61000L);        // 6. 验证状态恢复了（累计了第一条数据）        ConcurrentLinkedQueue<Object> output = testHarness.getOutput();        StreamRecord<UserProfile> record = (StreamRecord<UserProfile>) output.poll();        assertEquals(1, record.getValue().getClickCount());        assertEquals(99.9, record.getValue().getPurchaseAmount());    }    @Test    void testLateDataHandling() throws Exception {        // 先处理一个正常数据        UserAction normal = new UserAction("u1", ActionType.CLICK, "p1", "electronics", 0, 1000L);        testHarness.processElement(new StreamRecord<>(normal, 1000L));        // 推进水位线，关闭窗口        testHarness.setProcessingTime(70000L);        // 发送迟到数据（定时器已经触发）        UserAction late = new UserAction("u1", ActionType.PURCHASE, "p1", "electronics", 99.9, 2000L);        testHarness.processElement(new StreamRecord<>(late, 2000L));        // 验证迟到数据被丢弃或进入侧输出        // 根据业务逻辑，可能丢弃，不验证具体行为，只验证不崩溃        assertNotNull(testHarness.getOutput());    }}

为以下Flink SQL查询生成测试用例：SQL：[粘贴SQL]测试框架：Flink SqlValidator或FlinkMiniCluster测试数据（内联）：- source表数据：手动构造List<Row>- 期望结果：预期输出测试场景：1. 正常场景：验证聚合正确2. 边界场景：空输入，验证不输出3. 乱序场景：带水位线的测试要求：- 使用TableEnvironment- 使用registerDataStream或createTemporaryView- 使用executeSql执行查询- 使用collect()收集结果并断言生成完整的JUnit测试类。

package com.company.realtime.sql;import org.apache.flink.table.api.TableEnvironment;import org.apache.flink.table.api.TableResult;import org.apache.flink.types.Row;import org.junit.jupiter.api.Test;import java.util.List;import static org.junit.jupiter.api.Assertions.*;class ClickStatsSqlTest {    @Test    void testRollingWindowAggregation() throws Exception {        // 1. 创建TableEnvironment        TableEnvironment env = TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());        // 2. 注册源表（用VALUES模拟数据）        env.executeSql(            "CREATE TABLE click_log (" +            "  user_id STRING," +            "  product_id STRING," +            "  event_time TIMESTAMP(3)," +            "  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND" +            ") WITH (" +            "  'connector' = 'values'," +            "  'data-id' = 'test-data'" +            ")"        );        // 3. 插入测试数据（Flink 1.15+支持）        env.executeSql(            "INSERT INTO click_log VALUES " +            "('u1', 'p1', TIMESTAMP '2024-01-15 10:00:00')," +            "('u1', 'p1', TIMESTAMP '2024-01-15 10:00:10')," +            "('u2', 'p1', TIMESTAMP '2024-01-15 10:00:20')," +            "('u1', 'p2', TIMESTAMP '2024-01-15 10:00:30')"        );        // 4. 执行查询        TableResult result = env.executeSql(            "SELECT " +            "  product_id," +            "  TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start," +            "  COUNT(*) AS click_count," +            "  COUNT(DISTINCT user_id) AS uv_count " +            "FROM click_log " +            "GROUP BY product_id, TUMBLE(event_time, INTERVAL '1' MINUTE)"        );        // 5. 收集结果        List<Row> rows = result.collect().stream().toList();        // 6. 断言        assertEquals(2, rows.size()); // p1和p2各一个窗口        // 验证p1的统计        Row p1Row = rows.stream()            .filter(r -> r.getField(0).equals("p1"))            .findFirst()            .orElseThrow();        assertEquals(3L, p1Row.getField(2)); // click_count=3        assertEquals(2L, p1Row.getField(3)); // uv_count=2 (u1,u2)        // 验证p2的统计        Row p2Row = rows.stream()            .filter(r -> r.getField(0).equals("p2"))            .findFirst()            .orElseThrow();        assertEquals(1L, p2Row.getField(2)); // click_count=1        assertEquals(1L, p2Row.getField(3)); // uv_count=1    }}

为实时数据管道生成数据质量检查SQL：数据流：user_behavior_log（user_id, action_type, item_id, event_time）检查项：1. 完整性：user_id不为null2. 有效性：action_type在(click, purchase, favorite)内3. 时效性：event_time不晚于当前时间4. 一致性：item_id不为空且长度>05. 唯一性：同一window内相同user_id+item_id去重（可选）要求：- 生成Flink SQL查询，输出质量报告- 报告字段：check_name, total_count, invalid_count, invalid_rate- 使用侧输出流收集异常数据生成SQL。

-- 1. 完整性检查：user_id为nullSELECT     'null_user_id' AS check_name,    COUNT(*) AS total_count,    SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) AS invalid_count,    CAST(SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) AS invalid_rateFROM user_behavior_logUNION ALL-- 2. 有效性检查：action_type不在枚举中SELECT     'invalid_action_type' AS check_name,    COUNT(*) AS total_count,    SUM(CASE WHEN action_type NOT IN ('click', 'purchase', 'favorite') THEN 1 ELSE 0 END) AS invalid_count,    CAST(SUM(CASE WHEN action_type NOT IN ('click', 'purchase', 'favorite') THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) AS invalid_rateFROM user_behavior_logUNION ALL-- 3. 时效性检查：event_time是未来时间SELECT     'future_event_time' AS check_name,    COUNT(*) AS total_count,    SUM(CASE WHEN event_time > CURRENT_TIMESTAMP THEN 1 ELSE 0 END) AS invalid_count,    CAST(SUM(CASE WHEN event_time > CURRENT_TIMESTAMP THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) AS invalid_rateFROM user_behavior_logUNION ALL-- 4. 一致性检查：item_id为空SELECT     'empty_item_id' AS check_name,    COUNT(*) AS total_count,    SUM(CASE WHEN item_id IS NULL OR CHAR_LENGTH(item_id) = 0 THEN 1 ELSE 0 END) AS invalid_count,    CAST(SUM(CASE WHEN item_id IS NULL OR CHAR_LENGTH(item_id) = 0 THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) AS invalid_rateFROM user_behavior_log;

在Flink DataStream API中，实现数据质量过滤和侧输出：要求：1. 完整性检查：user_id为null → 侧输出2. 有效性检查：action_type不在枚举内 → 侧输出3. 时效性检查：event_time > 当前时间 → 侧输出（告警但保留数据）4. 正常数据：主输出定义3个侧输出流：- invalidNullSideOutput- invalidActionSideOutput- futureEventSideOutput生成完整的ProcessFunction代码。

为Flink作业生成压力测试脚本：作业：UserProfileJob（用户画像）集群：4个TaskManager，每个4核8G测试场景：1. 正常负载：10k条/秒2. 峰值负载：50k条/秒（持续5分钟）3. 反压测试：100k条/秒（持续1分钟，观察反压恢复）测试指标：- 端到端延迟（P50, P99, P999）- 吞吐量（records/s）- Checkpoint耗时和大小- 状态大小增长趋势- 资源使用（CPU/内存）要求：- 使用Apache JMeter或自定义Kafka生产者- 编写脚本：启动Flink作业，发送测试数据，收集metrics- 输出报告模板生成Shell脚本或Python脚本。

# flink_performance_test.pyimport timeimport jsonimport randomfrom kafka import KafkaProducerfrom prometheus_api_client import PrometheusConnectimport subprocessclass FlinkPerformanceTest:    def __init__(self):        self.producer = KafkaProducer(            bootstrap_servers=['localhost:9092'],            value_serializer=lambda v: json.dumps(v).encode('utf-8')        )        self.prometheus = PrometheusConnect(url="http://localhost:9090")    def test_normal_load(self):        """测试正常负载：10k/s"""        print("开始正常负载测试...")        start_time = time.time()        total_sent = 0        target_rate = 10000  # 10k/s        while time.time() - start_time < 300:  # 5分钟            batch_start = time.time()            sent_in_this_batch = 0            # 每秒发送目标数量            while time.time() - batch_start < 1 and sent_in_this_batch < target_rate:                self.send_random_event()                sent_in_this_batch += 1                total_sent += 1            # 等待到这一秒结束            time.sleep(max(0, 1 - (time.time() - batch_start)))        duration = time.time() - start_time        actual_rate = total_sent / duration        print(f"正常负载测试完成: 实际速率={actual_rate:.0f}/s")        # 收集metrics        self.collect_metrics()    def test_peak_load(self):        """测试峰值负载：50k/s，持续5分钟"""        print("开始峰值负载测试...")        # 实现类似逻辑，target_rate=50000    def test_backpressure_recovery(self):        """测试反压恢复"""        print("开始反压恢复测试...")        # 1. 先发送正常负载5分钟        self.send_load(10000, 300)        # 2. 突然增加到100k/s，持续1分钟        self.send_load(100000, 60)        # 3. 观察恢复时间        recovery_start = time.time()        self.send_load(10000, 300)        recovery_time = time.time() - recovery_start        print(f"反压恢复时间: {recovery_time:.1f}秒")    def collect_metrics(self):        """从Flink REST API收集指标"""        # 调用Flink REST API        result = subprocess.run(            ['curl', '-s', 'http://localhost:8081/jobs'],            capture_output=True, text=True        )        # 解析并记录...    def generate_report(self):        """生成测试报告"""        report = f"""        ===== Flink性能测试报告 =====        正常负载 (10k/s):          - 吞吐量: {self.normal_load_throughput:.0f} records/s          - 延迟 P99: {self.normal_load_latency_p99:.0f} ms          - Checkpoint大小: {self.normal_load_checkpoint_size:.2f} MB        峰值负载 (50k/s):          - 是否反压: {self.peak_load_backpressure}          - 延迟增加: {self.peak_load_latency_increase}x        反压恢复测试:          - 恢复时间: {self.recovery_time:.1f}秒        建议:          {self.get_recommendations()}        """        print(report)if __name__ == "__main__":    tester = FlinkPerformanceTest()    tester.test_normal_load()    tester.test_peak_load()    tester.test_backpressure_recovery()    tester.generate_report()

为以下Flink作业生成运维手册：[粘贴完整的Flink作业代码]要求：1. 作业概述：做什么、为什么2. 启动方式：命令行、参数说明3. 配置参数：所有可配置项及其含义4. 监控指标：关键指标及正常范围5. 常见问题：已知问题和解决方案6. 故障排查：日志关键字、排查步骤7. 扩缩容指南：如何调整并行度8. 止损方案：出现问题时的紧急操作输出格式：Markdown

# UserProfileJob 运维手册## 1. 作业概述**功能**：实时计算用户画像标签（短期5分钟滑动窗口 + 长期1小时定时器）**数据流**：Kafka(user-actions) → Flink → Redis(user:profile:*)**负责人**：实时数据团队## 2. 启动方式### 2.1 命令行启动```bash./bin/flink run \  -c com.company.realtime.profile.UserProfileJob \  -p 16 \  -yD taskmanager.memory.process.size=4096m \  -yD state.backend.rocksdb.incremental=true \  ./realtime-user-profile-1.0.0.jar \  --kafka.brokers kafka01:9092,kafka02:9092 \  --redis.host redis-cluster.prod

# Flink Web UIhttp://flink-ui:8081# 通过REST APIcurl http://flink-ui:8081/jobs/{jobid}/metrics?get=numRecordsInPerSecond

# 1. 从savepoint停止./bin/flink stop --savepointPath hdfs://savepoint $JOB_ID# 2. 以新并行度启动./bin/flink run -s hdfs://savepoint/savepoint-xxx -p 32 ...

# 取消作业（不保存savepoint）./bin/flink cancel $JOB_ID# 或者带savepoint停止./bin/flink stop --savepointPath hdfs://emergency $JOB_ID

### 7.5.2 生成架构图（Mermaid格式）**提示词**：

**Claude生成的Mermaid**：```mermaidgraph TD    A[Kafka: user-actions<br/>16 partitions] -->|JSON| B[Deserialize & Watermark<br/>并行度16]    B -->|WatermarkStrategy: 5s乱序| C[KeyBy user_id]    C -->|Split| D[ShortTerm Profile]    C -->|Split| E[LongTerm Profile]    subgraph D [短期画像 - 滑动窗口]        D1[窗口: 5分钟长度<br/>10秒步长]        D2[AggregateFunction<br/>增量聚合]        D3[WindowFunction<br/>输出画像]        D1 --> D2 --> D3    end    subgraph E [长期画像 - 定时器]        E1[ValueState累加器<br/>TTL=7天]        E2[定时器: 1分钟触发]        E3[输出累计画像]        E1 --> E2 --> E3    end    D3 --> F[Union]    E3 --> F    F -->|写入Redis| G[Redis: user:profile:*<br/>TTL: 短期5分钟/长期1小时]    B -.->|侧输出| H[Dead Letter Queue<br/>JSON解析失败]    style D fill:#f9f,stroke:#333    style E fill:#bbf,stroke:#333    style G fill:#bfb,stroke:#333

为UserProfileJob生成API文档：输出格式：Markdown，包含：1. 输入数据格式（Kafka消息Schema）2. 输出数据格式（Redis中的JSON）3. 错误码和异常4. 配置项说明要求：- 字段类型、说明、示例值- 必填/可选标识- 版本变更记录

.claude/├── prompts/│   ├── generate-flink-job.yaml      # 生成Flink作业的模板│   ├── generate-unit-test.yaml      # 生成单元测试的模板│   ├── generate-doc.yaml            # 生成文档的模板│   └── review-code.yaml             # Code Review模板├── rules/│   ├── code-style.yaml              # 代码规范│   ├── naming-convention.yaml       # 命名规范│   └── test-coverage.yaml           # 测试覆盖率要求└── knowledge/    ├── common-bugs.md               # 常见Bug库    ├── best-practices.md            # 最佳实践    └── faq.md                       # 常见问题

# 团队代码规范（AI会遵循）flink:  checkpoint_interval: 60000  checkpoint_mode: EXACTLY_ONCE  state_backend: RocksDB  restart_strategy: fixed-delay(3, 10s)java:  naming:    class: PascalCase    method: camelCase  annotations:    required: [@Override, @NonNull, @Nullable]  javadoc:    required_for: [public_class, public_method]testing:  coverage_threshold: 80  required_scenarios: [normal, boundary, error]

记录以下故障案例，加入知识库：【故障名称】：checkpoint-timeout-large-state【症状】：checkpoint耗时从30s增长到180s，最终超时失败【根因】：categoryPreference MapState没有限制大小，某些用户类目超过1000个【修复方案】：限制MapState只保留Top 10类目【相关代码】：LongTermProfileFunction.java 第87行【预防措施】：所有MapState必须设置大小限制请格式化后，写入.claude/knowledge/common-bugs.md

name: Flink CI with AI Reviewon:  pull_request:    paths:      - 'src/**/*.java'      - 'src/**/*.sql'jobs:  ai-review:    runs-on: ubuntu-latest    steps:      - uses: actions/checkout@v3      - name: Run AI Code Review        uses: anthropic/claude-code-action@v1        with:          prompt: |            Review this Flink code changes:            - Check for missing state TTL            - Check for potential data skew            - Check for unit test coverage            - Generate review report          api_key: ${{ secrets.CLAUDE_API_KEY }}      - name: Generate Documentation        run: |          claude-code --prompt "Generate updated documentation for changed files"      - name: Validate        run: mvn test

基于Git commit历史，生成Release Notes：commit范围：v1.2.0..HEAD筛选：只包含src/main/java下的修改输出格式：- 新增功能- 性能优化- Bug修复- 配置变更- 升级注意事项生成Markdown格式。