戰(zhàn))
EventBridge Pipes 完全仿真指南Floci 中的管道配置、Kafka 橋接與 Enrichment 實(shí)戰(zhàn)【免費(fèi)下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項(xiàng)目地址: https://gitcode.com/gh_mirrors/fl/flociFlociLight, fluffy, and always free — AWS 本地模擬器替代品提供了一套與 AWS EventBridge Pipes 線協(xié)議兼容的 REST-JSON API讓你在本地的 SQS、Kinesis、DynamoDB Streams、Kafka 等源與 Lambda、SQS、SNS、Kinesis、Step Functions 等目標(biāo)之間搭建可過(guò)濾、可富化Enrichment的事件管道。本指南以 docs/services/pipes.md 為主線結(jié)合 PipesController.java、PipesService.java、PipesPoller.java 等源碼幫助你掌握創(chuàng)建、描述、更新、啟停與刪除 Pipe 的全部操作理解 Kafka 源如何借助 Karapace 邊車工作以及 Enrichment、Filter、ParallelizationFactor 等參數(shù)的精確語(yǔ)義與當(dāng)前限制。概覽協(xié)議、端點(diǎn)與支持的動(dòng)作EventBridge Pipes 在 Floci 中通過(guò)REST-JSON協(xié)議暴露與需要X-Amz-Target的 JSON 1.1 協(xié)議不同固定端點(diǎn)為POST http://localhost:4566/從 PipesController.java 可以看到控制器使用標(biāo)準(zhǔn) HTTP 動(dòng)詞 JSON 請(qǐng)求體并統(tǒng)一以application/json生產(chǎn)/消費(fèi)。端點(diǎn)全部掛載在/v1/pipes下與 AWS CLI 的pipes子命令一一對(duì)應(yīng)。支持的動(dòng)作如下表ActionHTTP 路由說(shuō)明CreatePipePOST /v1/pipes/{name}以 source、target 與可選 enrichment 創(chuàng)建新的管道DescribePipeGET /v1/pipes/{name}獲取管道詳情含狀態(tài)與配置UpdatePipePUT /v1/pipes/{name}更新管道配置source、target、role、enrichment、desired state 等DeletePipeDELETE /v1/pipes/{name}刪除管道ListPipesGET /v1/pipes列出管道支持按狀態(tài)與前綴過(guò)濾StartPipePOST /v1/pipes/{name}/start啟動(dòng)已停止的管道StopPipePOST /v1/pipes/{name}/stop停止運(yùn)行中的管道請(qǐng)求體字段采用 AWS 線格式的大駝峰命名Source、Target、RoleArn、Description、Enrichment、DesiredState、SourceParameters、TargetParameters、EnrichmentParameters、Tags?？刂破鲗?duì)這些字段逐個(gè)解析并透?jìng)鹘o PipesService?？焖偕鲜諷QS 到 Lambda 的管道準(zhǔn)備環(huán)境先設(shè)置 AWS 端點(diǎn)環(huán)境變量指向本地運(yùn)行的 Flociexport AWS_ENDPOINT_URLhttp://localhost:4566創(chuàng)建管道SQS → Lambda# Create a pipe (SQS to Lambda) aws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --endpoint-url $AWS_ENDPOINT_URLCreatePipe有四個(gè)必填字段PipesService.createPipe 中會(huì)嚴(yán)格校驗(yàn)Name管道名稱不能為空Source源 ARN不能為空Target目標(biāo) ARN不能為空RoleArnIAM 角色 ARN不能為空。任一缺失都會(huì)返回ValidationExceptionHTTP 400。若同名管道已存在按region::name作為存儲(chǔ)鍵則返回ConflictExceptionHTTP 409。創(chuàng)建時(shí)若未指定DesiredState默認(rèn)值為RUNNING管道創(chuàng)建后會(huì)立即被 PipesPoller 調(diào)度輪詢。描述 / 列表 / 啟停 / 更新 / 刪除# Describe a pipe aws pipes describe-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # List all pipes aws pipes list-pipes \ --endpoint-url $AWS_ENDPOINT_URL # Start a pipe aws pipes start-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Stop a pipe aws pipes stop-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Update a pipe aws pipes update-pipe \ --name my-pipe \ --target arn:aws:lambda:us-east-1:000000000000:function:new-function \ --endpoint-url $AWS_ENDPOINT_URL # Delete a pipe aws pipes delete-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL幾個(gè)值得注意的細(xì)節(jié)describe-pipe返回完整的管道配置含SourceParameters、TargetParameters、EnrichmentParameters、Tags、CreationTime、LastModifiedTime等見 PipesController.buildPipeResponselist-pipes只返回摘要ARN、名稱、源、目標(biāo)、期望狀態(tài)、當(dāng)前狀態(tài)、時(shí)間戳不含SourceParameters與 AWS 行為一致見 buildPipeListEntrylist-pipes支持--name-prefix、--source-prefix、--target-prefix、--desired-state、--current-state過(guò)濾對(duì)應(yīng) PipesService.listPipesupdate-pipe采用“未提供的屬性保持原值”的語(yǔ)義只有顯式傳入的字段才會(huì)被覆蓋null 屬性不會(huì)清空既有值見 writePipeConfigurationdelete-pipe會(huì)先停止輪詢?cè)賱h除存儲(chǔ)記錄stop-pipe會(huì)取消輪詢定時(shí)器、清空 Kinesis/DynamoDB 迭代器并關(guān)閉 Kafka 消費(fèi)者見 PipesPoller.stopPolling。Floci 的pipes服務(wù)還實(shí)現(xiàn)了資源標(biāo)簽?zāi)芰agResource/untagResource/listTags并作為pipes:pipe資源類型接入 Resource Explorer見 PipesService.getSupportedResourceTypes。管道狀態(tài)機(jī)DesiredState與CurrentState兩個(gè)概念需要區(qū)分DesiredStateRUNNING或STOPPED表示期望狀態(tài)CurrentState實(shí)際狀態(tài)。PipeState 枚舉完整定義了以下狀態(tài)STARTING— 管道正在啟動(dòng)RUNNING— 管道正在活躍處理事件STOPPING— 管道正在停止STOPPED— 管道已停止不再處理事件DELETED— 管道已被刪除源碼中枚舉還包含CREATING、UPDATING、DELETING、CREATE_FAILED、UPDATE_FAILED、START_FAILED、STOP_FAILED、DELETE_FAILED等完整狀態(tài)集與 AWS 線格式保持一致。狀態(tài)流轉(zhuǎn)邏輯在 PipesService 中創(chuàng)建時(shí)未指定DesiredState則默認(rèn)為RUNNING此時(shí)CurrentState直接取RUNNING并啟動(dòng)輪詢createPipe創(chuàng)建時(shí)指定--desired-state STOPPEDCurrentState即為STOPPED不會(huì)啟動(dòng)輪詢start-pipe將 DesiredState 與 CurrentState 都置為RUNNING并恢復(fù)輪詢startPipestop-pipe先停止輪詢?cè)賹蓚€(gè)狀態(tài)置為STOPPEDstopPipe。兼容性測(cè)試 pipes.bats 中驗(yàn)證了--desired-state STOPPED創(chuàng)建后CurrentState STOPPED的行為。另外Floci 重啟后會(huì)自動(dòng)恢復(fù)CurrentState RUNNING的管道輪詢startPersistedPollers保證模擬環(huán)境中斷后管道能繼續(xù)消費(fèi)。支持的源與目標(biāo)Floci 模擬的 EventBridge Pipes 支持以下源與目標(biāo)類型由 PipesPoller.pollAndInvoke 依據(jù)源 ARN 中的服務(wù)段自動(dòng)分派Sources源Amazon SQS 隊(duì)列arn:aws:sqs:...Amazon Kinesis 流arn:aws:kinesis:...Amazon DynamoDB 流arn:aws:dynamodb:...Kafka topics —— MSKarn:aws:kafka:...與自管理smk://協(xié)議Targets目標(biāo)Lambda 函數(shù):lambda:或:function:SQS 隊(duì)列SNS 主題Kinesis 流Step Functions 狀態(tài)機(jī):states:目標(biāo)分派邏輯在 PipesTargetInvoker.invoke按 ARN 特征路由到對(duì)應(yīng)服務(wù)Lambda以RequestResponse同步調(diào)用若返回FunctionError視為投遞失敗并轉(zhuǎn)入 DLQinvokeLambdaSQS將 payload 作為消息體發(fā)送到隊(duì)列SNS以SubjectPipes發(fā)布到主題invokeSnsEventBridge向事件總線PutEvents默認(rèn)Sourceaws.pipes、DetailTypePipeForwarded可通過(guò)EventBridgeEventBusParameters覆蓋invokeEventBridgeStep Functions以隨機(jī)執(zhí)行名pipes-uuid啟動(dòng)執(zhí)行invokeStepFunctions。目標(biāo)側(cè)還支持TargetParameters.InputTemplate模板中用$.path形式的 JSONPath 占位符投遞前替換為從事件中提取的值實(shí)現(xiàn)在 applyInputTemplate / extractJsonPath。Kafka 源與 Karapace REST 橋接Kafka 源MSK 或自管理smk://是 Floci Pipes 中最特殊的一條路徑它按需啟動(dòng)一個(gè) Karapace REST Proxy 邊車容器通過(guò) HTTP 輪詢 Kafka而不是在 Floci 進(jìn)程內(nèi)嵌入kafka-clients參見 KarapaceManager 的注釋關(guān)聯(lián) issue #2916。工作原理每個(gè)不同的bootstrap.servers目標(biāo)對(duì)應(yīng)一個(gè) Karapace 容器多個(gè)讀取同一集群的 Pipe共享這一個(gè)實(shí)例首次使用時(shí)ensureStarted按需啟動(dòng)通過(guò) SHA-256 摘要的前 8 個(gè)十六進(jìn)制字符生成目標(biāo)鍵并用該鍵構(gòu)造容器名pipes-kafka-rest-hash容器以KARAPACE_*環(huán)境變量配置KARAPACE_HOST0.0.0.0、KARAPACE_PORT8082、KARAPACE_BOOTSTRAP_URIbootstrap、KARAPACE_ADVERTISED_PROTOCOLhttp并顯式指定入口/venv/bin/karapace_rest_proxy鏡像默認(rèn)入口是無(wú)操作的python3見 KarapaceManager.startContainer啟動(dòng)后通過(guò)GET /_health探測(cè)就緒30 秒超時(shí)隨后 Floci 的 PipesKafkaRestClient 使用 Kafka REST Proxy v2 API 子集消費(fèi)并提交偏移量地址改寫若bootstrap.servers是localhost/127.0.0.1會(huì)改寫為 Docker 主機(jī)可達(dá)地址host.docker.internal別名并在 Linux 原生 Docker 上顯式注入 host-gateway 映射見 resolveForContainer進(jìn)程退出時(shí)PreDestroy統(tǒng)一停止并刪除全部 Karapace 容器。相關(guān)配置環(huán)境變量默認(rèn)值說(shuō)明FLOCI_SERVICES_PIPES_ENABLEDtrue啟用或禁用該服務(wù)FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_DEFAULT_IMAGEghcr.io/aiven-open/karapace:latestKafka 源管道按需啟動(dòng)的 Karapace REST Proxy 邊車鏡像FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_BASE9500分配給 Karapace 邊車的主機(jī)端口段起點(diǎn)FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_MAX9599分配給 Karapace 邊車的主機(jī)端口段終點(diǎn)配置實(shí)現(xiàn)在 EmulatorConfig.PipesServiceConfig / KafkaRestBridgeConfigKafkaRestBridgeConfig定義了默認(rèn)鏡像與端口段KarapaceManager 通過(guò)PortAllocator在[9500, 9599]區(qū)間分配主機(jī)端口容器內(nèi)運(yùn)行時(shí)不綁定主機(jī)端口僅暴露 8082。重要前提帶 Kafka 源的管道依賴 Docker 環(huán)境因?yàn)樾枰獑?dòng) Karapace 邊車其余源SQS/Kinesis/DynamoDB Streams不需要 Docker。Kafka 源的參數(shù)要求創(chuàng)建 Kafka 源管道時(shí)SourceParameters必須攜帶對(duì)應(yīng)的參數(shù)塊否則返回ValidationException見 validateSourceConfigurationsmk://前綴 → 必須提供SelfManagedKafkaParameters:kafka:ARN → 必須提供ManagedStreamingKafkaParameters且參數(shù)塊內(nèi)TopicName為必填項(xiàng)。示例aws pipes create-pipe \ --name my-kafka-pipe \ --source arn:aws:kafka:us-east-1:000000000000:cluster/my-msk/b4f3c2a1-0000-4000-8000-000000000000-9 \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { ManagedStreamingKafkaParameters: { TopicName: orders, StartingPosition: TRIM_HORIZON } } \ --endpoint-url $AWS_ENDPOINT_URL自管理 Kafka 的smk://源同理但使用SelfManagedKafkaParameters并在其中給出BootstrapServers。ParallelizationFactor參數(shù)語(yǔ)義與當(dāng)前限制CreatePipe與UpdatePipe在KinesisStreamParameters和DynamoDBStreamParameters源參數(shù)塊中接受1 到 10 之間的整數(shù)ParallelizationFactor與 AWS 線格式一致。DescribePipe會(huì)在SourceParameters中原樣回顯ListPipes只返回摘要、不含SourceParameters與 AWS 一致。aws pipes create-pipe \ --name kinesis-pipe \ --source arn:aws:kinesis:us-east-1:000000000000:stream/events \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters {KinesisStreamParameters:{StartingPosition:TRIM_HORIZON,ParallelizationFactor:4}} \ --endpoint-url $AWS_ENDPOINT_URL校驗(yàn)邏輯嚴(yán)格對(duì)齊 AWSvalidateParallelizationFactor值超出 1~10 →ValidationException字段出現(xiàn)在與源不匹配的參數(shù)塊上例如 SQS 源攜帶KinesisStreamParameters.ParallelizationFactor→ValidationException非數(shù)字、非整數(shù)、超出 int 范圍的寬值如2^64 5也會(huì)被拒絕——源碼通過(guò)canConvertToInt()防止 BigInteger 溢出繞過(guò)邊界檢查。Enforcement status重要限制ParallelizationFactor目前僅持久化并在線格式上回顯輪詢器尚未按分片并發(fā)處理批次。Floci 對(duì)每個(gè) Kinesis / DynamoDB Streams 管道只在單個(gè)分片shardId-000000000000上打開迭代器、每次只投遞一個(gè)批次無(wú)論配置值是多少見 initKinesisIterator 與 initDynamoDbIterator。多分片輪詢與真正的分片級(jí)并發(fā)被列為后續(xù)跟進(jìn)項(xiàng)。Filter事件過(guò)濾與匹配語(yǔ)義管道支持 AWS 風(fēng)格的FilterCriteria過(guò)濾位于SourceParameters.FilterCriteria.Filters每個(gè)過(guò)濾條件包含PatternJSON 匹配模式。實(shí)現(xiàn)位于 PipesFilterMatcher多條 filter 之間是OR關(guān)系任一匹配即通過(guò)單條 filter 內(nèi)的多個(gè)字段是AND關(guān)系模式值必須是數(shù)組或?qū)ο髽?biāo)量模式會(huì)被判定為非法并拒絕該記錄支持字符串精確匹配、數(shù)字精確匹配、null字段不存在、對(duì)象嵌套支持 AWS 特殊模式{prefix: ...}、{suffix: ...}、{equals-ignore-case: ...}、{anything-but: ...}含anything-butprefix、{exists: true|false}、{numeric: [, 5, , 1]}等。SQS 路徑下未匹配過(guò)濾條件的消息會(huì)被立即從源隊(duì)列刪除AWS 行為被過(guò)濾掉的消息視為已消費(fèi)見 PipesPoller.pollSqs。Kafka 路徑下未匹配的記錄會(huì)推進(jìn)提交偏移量相當(dāng)于消費(fèi)掉見 deliverKafkaRecords。EnrichmentLambda 富化與 DLQ 兜底管道的可選富化步驟source → filter → enrichment → target目前僅為 Lambda enrichment 提供模擬過(guò)濾后的批次被同步調(diào)用RequestResponse富化函數(shù)的響應(yīng)成為目標(biāo)的輸入。實(shí)現(xiàn)在 PipesTargetInvoker.applyEnrichment。行為細(xì)則與 AWS 對(duì)齊空響應(yīng)跳過(guò)目標(biāo)空 body、null、{}、[]都會(huì)消費(fèi)源記錄而不調(diào)用目標(biāo)非空數(shù)組如[{}]仍會(huì)調(diào)用目標(biāo)攜帶一個(gè)空 payload 元素。注意源碼會(huì)先解析 JSON 以區(qū)分{ }、[ ]等空白變體Lambda 富化返回FunctionError會(huì)判定批次失敗源記錄被路由到管道的死信隊(duì)列DLQ而非靜默消費(fèi)非 Lambda 富化類型API destinations、API Gateway、Step Functions Express在 AWS 上合法但 Floci不模擬配置了此類富化的管道會(huì)把批次判失敗并轉(zhuǎn)入 DLQ而不是投遞未富化的 payload富化目前只在 SQS 源路徑生效Kinesis、DynamoDB Streams 與 Kafka 源會(huì)把過(guò)濾后的記錄直接投遞給目標(biāo)配置在這些源上的 enrichment 暫不生效見 PipesPoller.deliverRecords 的注釋。富化響應(yīng)到目標(biāo)的形狀轉(zhuǎn)換富化返回的是單個(gè) JSON 值非數(shù)組時(shí)代表一條事件會(huì)被包裝成單元素?cái)?shù)組返回?cái)?shù)組則視為已是批次、原樣轉(zhuǎn)發(fā)。該邏輯在 asEventArray。同時(shí)Lambda 目標(biāo)接收的輸入保持{Records: [...]}批次形狀而 Step Functions、SQS、SNS、EventBridge 目標(biāo)接收富化響應(yīng)的原始內(nèi)容數(shù)組包裝會(huì)破壞它們的輸入見 deliverEnrichedBatch。DLQ死信隊(duì)列配置管道的 DLQ 通過(guò)源參數(shù)塊中的DeadLetterConfig.Arn配置支持SqsQueueParameters、KinesisStreamParameters、DynamoDBStreamParameters三個(gè)塊getDlqArnaws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { SqsQueueParameters: { BatchSize: 10, DeadLetterConfig: {Arn: arn:aws:sqs:us-east-1:000000000000:dlq-queue} } } \ --endpoint-url $AWS_ENDPOINT_URL投遞失敗時(shí)sendToDeadLetterQueue 會(huì)把失敗 payload 發(fā)送到 DLQ若未配置 DLQ則返回投遞失敗計(jì)數(shù)并記錄告警日志。批次大小與輪詢行為默認(rèn)BatchSize為 10可通過(guò)各源參數(shù)塊的BatchSize字段調(diào)整getBatchSize輪詢間隔固定為 1 秒POLL_INTERVAL_MS 1000通過(guò) Vert.x 周期定時(shí)器驅(qū)動(dòng)并使用獨(dú)立線程池執(zhí)行保證同一管道不會(huì)并發(fā)輪詢activePolls防重入見 PipesPollerKinesis 迭代器過(guò)期ExpiredIteratorException或 DynamoDB 流數(shù)據(jù)被裁剪TrimmedDataAccessException時(shí)會(huì)自動(dòng)重置迭代器并重新拉取pollKinesis、pollDynamoDbStreamsKafka 路徑下REST Proxy 一次輪詢可能返回超過(guò)一個(gè)BatchSize的記錄因此一個(gè)輪詢周期會(huì)順序投遞多個(gè)批次一旦某分區(qū)投遞失敗該分區(qū)在后續(xù)批次中也不會(huì)提交偏移量保證“先失敗記錄重投”的順序語(yǔ)義deliverKafkaBatch。與 CloudFormation 的集成Pipes 服務(wù)實(shí)現(xiàn)了ResourceProvider接口支持 CloudFormation 對(duì)AWS::Pipes::Pipe資源的創(chuàng)建、更新與刪除提供器位于 cloudformation/provisioners 下的PipesCfnProvisioner并配套了回滾恢復(fù)測(cè)試 PipesCfnRollbackRestoresExactConfigurationTest.java 與 PipesCfnProvisionerTest.java。restorePipe路徑采用“null 即清空”的全量恢復(fù)語(yǔ)義確保 CloudFormation 回滾時(shí)能把管道精確還原到快照狀態(tài)PipesService.restorePipe。測(cè)試與驗(yàn)證倉(cāng)庫(kù)提供了多層測(cè)試覆蓋 pipes 服務(wù)單元測(cè)試PipesServiceTest.java、PipesPollerTest.java、PipesTargetInvokerTest.java、PipesFilterMatcherTest.java、KarapaceManagerTest.java、PipesKafkaRestClientTest.java集成測(cè)試PipesIntegrationTest.java、PipesPollerIntegrationTest.javaAWS CLI 兼容性測(cè)試pipes.bats 覆蓋 create/describe/list/update/start/stop 全流程SDK 各語(yǔ)言套件sdk-test-go、sdk-test-node、sdk-test-java、sdk-test-python中的pipes用例驗(yàn)證多語(yǔ)言 SDK 兼容性。結(jié)語(yǔ)Floci 的 EventBridge Pipes 實(shí)現(xiàn)覆蓋了從創(chuàng)建、啟停、更新、刪除到列表過(guò)濾的完整生命周期并對(duì) SQS/Kinesis/DynamoDB Streams/Kafka 四種源與 Lambda/SQS/SNS/Kinesis/Step Functions/EventBridge 多種目標(biāo)提供事件搬運(yùn)能力Kafka 源通過(guò)按需啟動(dòng)的 Karapace REST 邊車實(shí)現(xiàn)零內(nèi)嵌依賴的輪詢。在使用時(shí)請(qǐng)?zhí)貏e注意當(dāng)前仿真邊界ParallelizationFactor僅為線格式兼容單分片順序輪詢、enrichment 僅支持 Lambda 且僅作用于 SQS 源路徑、Kafka 源依賴 Docker。這些限制都以代碼注釋和文檔!!! note塊的形式明確標(biāo)注便于你在搭建本地事件管道時(shí)做出準(zhǔn)確預(yù)期。【免費(fèi)下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項(xiàng)目地址: https://gitcode.com/gh_mirrors/fl/floci創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 EventBridge Pipes 完全仿真指南：Floci 中的管道配置、Kafka 橋接與 Enrichment 實(shí)戰(zhàn)
EventBridge Pipes 完全仿真指南Floci 中的管道配置、Kafka 橋接與 Enrichment 實(shí)戰(zhàn)【免費(fèi)下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項(xiàng)目地址: https://gitcode.com/gh_mirrors/fl/flociFlociLight, fluffy, and always free — AWS 本地模擬器替代品提供了一套與 AWS EventBridge Pipes 線協(xié)議兼容的 REST-JSON API讓你在本地的 SQS、Kinesis、DynamoDB Streams、Kafka 等源與 Lambda、SQS、SNS、Kinesis、Step Functions 等目標(biāo)之間搭建可過(guò)濾、可富化Enrichment的事件管道。本指南以 docs/services/pipes.md 為主線結(jié)合 PipesController.java、PipesService.java、PipesPoller.java 等源碼幫助你掌握創(chuàng)建、描述、更新、啟停與刪除 Pipe 的全部操作理解 Kafka 源如何借助 Karapace 邊車工作以及 Enrichment、Filter、ParallelizationFactor 等參數(shù)的精確語(yǔ)義與當(dāng)前限制。概覽協(xié)議、端點(diǎn)與支持的動(dòng)作EventBridge Pipes 在 Floci 中通過(guò)REST-JSON協(xié)議暴露與需要X-Amz-Target的 JSON 1.1 協(xié)議不同固定端點(diǎn)為POST http://localhost:4566/從 PipesController.java 可以看到控制器使用標(biāo)準(zhǔn) HTTP 動(dòng)詞 JSON 請(qǐng)求體并統(tǒng)一以application/json生產(chǎn)/消費(fèi)。端點(diǎn)全部掛載在/v1/pipes下與 AWS CLI 的pipes子命令一一對(duì)應(yīng)。支持的動(dòng)作如下表ActionHTTP 路由說(shuō)明CreatePipePOST /v1/pipes/{name}以 source、target 與可選 enrichment 創(chuàng)建新的管道DescribePipeGET /v1/pipes/{name}獲取管道詳情含狀態(tài)與配置UpdatePipePUT /v1/pipes/{name}更新管道配置source、target、role、enrichment、desired state 等DeletePipeDELETE /v1/pipes/{name}刪除管道ListPipesGET /v1/pipes列出管道支持按狀態(tài)與前綴過(guò)濾StartPipePOST /v1/pipes/{name}/start啟動(dòng)已停止的管道StopPipePOST /v1/pipes/{name}/stop停止運(yùn)行中的管道請(qǐng)求體字段采用 AWS 線格式的大駝峰命名Source、Target、RoleArn、Description、Enrichment、DesiredState、SourceParameters、TargetParameters、EnrichmentParameters、Tags?？刂破鲗?duì)這些字段逐個(gè)解析并透?jìng)鹘o PipesService?？焖偕鲜諷QS 到 Lambda 的管道準(zhǔn)備環(huán)境先設(shè)置 AWS 端點(diǎn)環(huán)境變量指向本地運(yùn)行的 Flociexport AWS_ENDPOINT_URLhttp://localhost:4566創(chuàng)建管道SQS → Lambda# Create a pipe (SQS to Lambda) aws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --endpoint-url $AWS_ENDPOINT_URLCreatePipe有四個(gè)必填字段PipesService.createPipe 中會(huì)嚴(yán)格校驗(yàn)Name管道名稱不能為空Source源 ARN不能為空Target目標(biāo) ARN不能為空RoleArnIAM 角色 ARN不能為空。任一缺失都會(huì)返回ValidationExceptionHTTP 400。若同名管道已存在按region::name作為存儲(chǔ)鍵則返回ConflictExceptionHTTP 409。創(chuàng)建時(shí)若未指定DesiredState默認(rèn)值為RUNNING管道創(chuàng)建后會(huì)立即被 PipesPoller 調(diào)度輪詢。描述 / 列表 / 啟停 / 更新 / 刪除# Describe a pipe aws pipes describe-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # List all pipes aws pipes list-pipes \ --endpoint-url $AWS_ENDPOINT_URL # Start a pipe aws pipes start-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Stop a pipe aws pipes stop-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL # Update a pipe aws pipes update-pipe \ --name my-pipe \ --target arn:aws:lambda:us-east-1:000000000000:function:new-function \ --endpoint-url $AWS_ENDPOINT_URL # Delete a pipe aws pipes delete-pipe \ --name my-pipe \ --endpoint-url $AWS_ENDPOINT_URL幾個(gè)值得注意的細(xì)節(jié)describe-pipe返回完整的管道配置含SourceParameters、TargetParameters、EnrichmentParameters、Tags、CreationTime、LastModifiedTime等見 PipesController.buildPipeResponselist-pipes只返回摘要ARN、名稱、源、目標(biāo)、期望狀態(tài)、當(dāng)前狀態(tài)、時(shí)間戳不含SourceParameters與 AWS 行為一致見 buildPipeListEntrylist-pipes支持--name-prefix、--source-prefix、--target-prefix、--desired-state、--current-state過(guò)濾對(duì)應(yīng) PipesService.listPipesupdate-pipe采用“未提供的屬性保持原值”的語(yǔ)義只有顯式傳入的字段才會(huì)被覆蓋null 屬性不會(huì)清空既有值見 writePipeConfigurationdelete-pipe會(huì)先停止輪詢?cè)賱h除存儲(chǔ)記錄stop-pipe會(huì)取消輪詢定時(shí)器、清空 Kinesis/DynamoDB 迭代器并關(guān)閉 Kafka 消費(fèi)者見 PipesPoller.stopPolling。Floci 的pipes服務(wù)還實(shí)現(xiàn)了資源標(biāo)簽?zāi)芰agResource/untagResource/listTags并作為pipes:pipe資源類型接入 Resource Explorer見 PipesService.getSupportedResourceTypes。管道狀態(tài)機(jī)DesiredState與CurrentState兩個(gè)概念需要區(qū)分DesiredStateRUNNING或STOPPED表示期望狀態(tài)CurrentState實(shí)際狀態(tài)。PipeState 枚舉完整定義了以下狀態(tài)STARTING— 管道正在啟動(dòng)RUNNING— 管道正在活躍處理事件STOPPING— 管道正在停止STOPPED— 管道已停止不再處理事件DELETED— 管道已被刪除源碼中枚舉還包含CREATING、UPDATING、DELETING、CREATE_FAILED、UPDATE_FAILED、START_FAILED、STOP_FAILED、DELETE_FAILED等完整狀態(tài)集與 AWS 線格式保持一致。狀態(tài)流轉(zhuǎn)邏輯在 PipesService 中創(chuàng)建時(shí)未指定DesiredState則默認(rèn)為RUNNING此時(shí)CurrentState直接取RUNNING并啟動(dòng)輪詢createPipe創(chuàng)建時(shí)指定--desired-state STOPPEDCurrentState即為STOPPED不會(huì)啟動(dòng)輪詢start-pipe將 DesiredState 與 CurrentState 都置為RUNNING并恢復(fù)輪詢startPipestop-pipe先停止輪詢?cè)賹蓚€(gè)狀態(tài)置為STOPPEDstopPipe。兼容性測(cè)試 pipes.bats 中驗(yàn)證了--desired-state STOPPED創(chuàng)建后CurrentState STOPPED的行為。另外Floci 重啟后會(huì)自動(dòng)恢復(fù)CurrentState RUNNING的管道輪詢startPersistedPollers保證模擬環(huán)境中斷后管道能繼續(xù)消費(fèi)。支持的源與目標(biāo)Floci 模擬的 EventBridge Pipes 支持以下源與目標(biāo)類型由 PipesPoller.pollAndInvoke 依據(jù)源 ARN 中的服務(wù)段自動(dòng)分派Sources源Amazon SQS 隊(duì)列arn:aws:sqs:...Amazon Kinesis 流arn:aws:kinesis:...Amazon DynamoDB 流arn:aws:dynamodb:...Kafka topics —— MSKarn:aws:kafka:...與自管理smk://協(xié)議Targets目標(biāo)Lambda 函數(shù):lambda:或:function:SQS 隊(duì)列SNS 主題Kinesis 流Step Functions 狀態(tài)機(jī):states:目標(biāo)分派邏輯在 PipesTargetInvoker.invoke按 ARN 特征路由到對(duì)應(yīng)服務(wù)Lambda以RequestResponse同步調(diào)用若返回FunctionError視為投遞失敗并轉(zhuǎn)入 DLQinvokeLambdaSQS將 payload 作為消息體發(fā)送到隊(duì)列SNS以SubjectPipes發(fā)布到主題invokeSnsEventBridge向事件總線PutEvents默認(rèn)Sourceaws.pipes、DetailTypePipeForwarded可通過(guò)EventBridgeEventBusParameters覆蓋invokeEventBridgeStep Functions以隨機(jī)執(zhí)行名pipes-uuid啟動(dòng)執(zhí)行invokeStepFunctions。目標(biāo)側(cè)還支持TargetParameters.InputTemplate模板中用$.path形式的 JSONPath 占位符投遞前替換為從事件中提取的值實(shí)現(xiàn)在 applyInputTemplate / extractJsonPath。Kafka 源與 Karapace REST 橋接Kafka 源MSK 或自管理smk://是 Floci Pipes 中最特殊的一條路徑它按需啟動(dòng)一個(gè) Karapace REST Proxy 邊車容器通過(guò) HTTP 輪詢 Kafka而不是在 Floci 進(jìn)程內(nèi)嵌入kafka-clients參見 KarapaceManager 的注釋關(guān)聯(lián) issue #2916。工作原理每個(gè)不同的bootstrap.servers目標(biāo)對(duì)應(yīng)一個(gè) Karapace 容器多個(gè)讀取同一集群的 Pipe共享這一個(gè)實(shí)例首次使用時(shí)ensureStarted按需啟動(dòng)通過(guò) SHA-256 摘要的前 8 個(gè)十六進(jìn)制字符生成目標(biāo)鍵并用該鍵構(gòu)造容器名pipes-kafka-rest-hash容器以KARAPACE_*環(huán)境變量配置KARAPACE_HOST0.0.0.0、KARAPACE_PORT8082、KARAPACE_BOOTSTRAP_URIbootstrap、KARAPACE_ADVERTISED_PROTOCOLhttp并顯式指定入口/venv/bin/karapace_rest_proxy鏡像默認(rèn)入口是無(wú)操作的python3見 KarapaceManager.startContainer啟動(dòng)后通過(guò)GET /_health探測(cè)就緒30 秒超時(shí)隨后 Floci 的 PipesKafkaRestClient 使用 Kafka REST Proxy v2 API 子集消費(fèi)并提交偏移量地址改寫若bootstrap.servers是localhost/127.0.0.1會(huì)改寫為 Docker 主機(jī)可達(dá)地址host.docker.internal別名并在 Linux 原生 Docker 上顯式注入 host-gateway 映射見 resolveForContainer進(jìn)程退出時(shí)PreDestroy統(tǒng)一停止并刪除全部 Karapace 容器。相關(guān)配置環(huán)境變量默認(rèn)值說(shuō)明FLOCI_SERVICES_PIPES_ENABLEDtrue啟用或禁用該服務(wù)FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_DEFAULT_IMAGEghcr.io/aiven-open/karapace:latestKafka 源管道按需啟動(dòng)的 Karapace REST Proxy 邊車鏡像FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_BASE9500分配給 Karapace 邊車的主機(jī)端口段起點(diǎn)FLOCI_SERVICES_PIPES_KAFKA_REST_BRIDGE_HOST_PORT_MAX9599分配給 Karapace 邊車的主機(jī)端口段終點(diǎn)配置實(shí)現(xiàn)在 EmulatorConfig.PipesServiceConfig / KafkaRestBridgeConfigKafkaRestBridgeConfig定義了默認(rèn)鏡像與端口段KarapaceManager 通過(guò)PortAllocator在[9500, 9599]區(qū)間分配主機(jī)端口容器內(nèi)運(yùn)行時(shí)不綁定主機(jī)端口僅暴露 8082。重要前提帶 Kafka 源的管道依賴 Docker 環(huán)境因?yàn)樾枰獑?dòng) Karapace 邊車其余源SQS/Kinesis/DynamoDB Streams不需要 Docker。Kafka 源的參數(shù)要求創(chuàng)建 Kafka 源管道時(shí)SourceParameters必須攜帶對(duì)應(yīng)的參數(shù)塊否則返回ValidationException見 validateSourceConfigurationsmk://前綴 → 必須提供SelfManagedKafkaParameters:kafka:ARN → 必須提供ManagedStreamingKafkaParameters且參數(shù)塊內(nèi)TopicName為必填項(xiàng)。示例aws pipes create-pipe \ --name my-kafka-pipe \ --source arn:aws:kafka:us-east-1:000000000000:cluster/my-msk/b4f3c2a1-0000-4000-8000-000000000000-9 \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { ManagedStreamingKafkaParameters: { TopicName: orders, StartingPosition: TRIM_HORIZON } } \ --endpoint-url $AWS_ENDPOINT_URL自管理 Kafka 的smk://源同理但使用SelfManagedKafkaParameters并在其中給出BootstrapServers。ParallelizationFactor參數(shù)語(yǔ)義與當(dāng)前限制CreatePipe與UpdatePipe在KinesisStreamParameters和DynamoDBStreamParameters源參數(shù)塊中接受1 到 10 之間的整數(shù)ParallelizationFactor與 AWS 線格式一致。DescribePipe會(huì)在SourceParameters中原樣回顯ListPipes只返回摘要、不含SourceParameters與 AWS 一致。aws pipes create-pipe \ --name kinesis-pipe \ --source arn:aws:kinesis:us-east-1:000000000000:stream/events \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters {KinesisStreamParameters:{StartingPosition:TRIM_HORIZON,ParallelizationFactor:4}} \ --endpoint-url $AWS_ENDPOINT_URL校驗(yàn)邏輯嚴(yán)格對(duì)齊 AWSvalidateParallelizationFactor值超出 1~10 →ValidationException字段出現(xiàn)在與源不匹配的參數(shù)塊上例如 SQS 源攜帶KinesisStreamParameters.ParallelizationFactor→ValidationException非數(shù)字、非整數(shù)、超出 int 范圍的寬值如2^64 5也會(huì)被拒絕——源碼通過(guò)canConvertToInt()防止 BigInteger 溢出繞過(guò)邊界檢查。Enforcement status重要限制ParallelizationFactor目前僅持久化并在線格式上回顯輪詢器尚未按分片并發(fā)處理批次。Floci 對(duì)每個(gè) Kinesis / DynamoDB Streams 管道只在單個(gè)分片shardId-000000000000上打開迭代器、每次只投遞一個(gè)批次無(wú)論配置值是多少見 initKinesisIterator 與 initDynamoDbIterator。多分片輪詢與真正的分片級(jí)并發(fā)被列為后續(xù)跟進(jìn)項(xiàng)。Filter事件過(guò)濾與匹配語(yǔ)義管道支持 AWS 風(fēng)格的FilterCriteria過(guò)濾位于SourceParameters.FilterCriteria.Filters每個(gè)過(guò)濾條件包含PatternJSON 匹配模式。實(shí)現(xiàn)位于 PipesFilterMatcher多條 filter 之間是OR關(guān)系任一匹配即通過(guò)單條 filter 內(nèi)的多個(gè)字段是AND關(guān)系模式值必須是數(shù)組或?qū)ο髽?biāo)量模式會(huì)被判定為非法并拒絕該記錄支持字符串精確匹配、數(shù)字精確匹配、null字段不存在、對(duì)象嵌套支持 AWS 特殊模式{prefix: ...}、{suffix: ...}、{equals-ignore-case: ...}、{anything-but: ...}含anything-butprefix、{exists: true|false}、{numeric: [, 5, , 1]}等。SQS 路徑下未匹配過(guò)濾條件的消息會(huì)被立即從源隊(duì)列刪除AWS 行為被過(guò)濾掉的消息視為已消費(fèi)見 PipesPoller.pollSqs。Kafka 路徑下未匹配的記錄會(huì)推進(jìn)提交偏移量相當(dāng)于消費(fèi)掉見 deliverKafkaRecords。EnrichmentLambda 富化與 DLQ 兜底管道的可選富化步驟source → filter → enrichment → target目前僅為 Lambda enrichment 提供模擬過(guò)濾后的批次被同步調(diào)用RequestResponse富化函數(shù)的響應(yīng)成為目標(biāo)的輸入。實(shí)現(xiàn)在 PipesTargetInvoker.applyEnrichment。行為細(xì)則與 AWS 對(duì)齊空響應(yīng)跳過(guò)目標(biāo)空 body、null、{}、[]都會(huì)消費(fèi)源記錄而不調(diào)用目標(biāo)非空數(shù)組如[{}]仍會(huì)調(diào)用目標(biāo)攜帶一個(gè)空 payload 元素。注意源碼會(huì)先解析 JSON 以區(qū)分{ }、[ ]等空白變體Lambda 富化返回FunctionError會(huì)判定批次失敗源記錄被路由到管道的死信隊(duì)列DLQ而非靜默消費(fèi)非 Lambda 富化類型API destinations、API Gateway、Step Functions Express在 AWS 上合法但 Floci不模擬配置了此類富化的管道會(huì)把批次判失敗并轉(zhuǎn)入 DLQ而不是投遞未富化的 payload富化目前只在 SQS 源路徑生效Kinesis、DynamoDB Streams 與 Kafka 源會(huì)把過(guò)濾后的記錄直接投遞給目標(biāo)配置在這些源上的 enrichment 暫不生效見 PipesPoller.deliverRecords 的注釋。富化響應(yīng)到目標(biāo)的形狀轉(zhuǎn)換富化返回的是單個(gè) JSON 值非數(shù)組時(shí)代表一條事件會(huì)被包裝成單元素?cái)?shù)組返回?cái)?shù)組則視為已是批次、原樣轉(zhuǎn)發(fā)。該邏輯在 asEventArray。同時(shí)Lambda 目標(biāo)接收的輸入保持{Records: [...]}批次形狀而 Step Functions、SQS、SNS、EventBridge 目標(biāo)接收富化響應(yīng)的原始內(nèi)容數(shù)組包裝會(huì)破壞它們的輸入見 deliverEnrichedBatch。DLQ死信隊(duì)列配置管道的 DLQ 通過(guò)源參數(shù)塊中的DeadLetterConfig.Arn配置支持SqsQueueParameters、KinesisStreamParameters、DynamoDBStreamParameters三個(gè)塊getDlqArnaws pipes create-pipe \ --name my-pipe \ --source arn:aws:sqs:us-east-1:000000000000:source-queue \ --target arn:aws:lambda:us-east-1:000000000000:function:my-function \ --role-arn arn:aws:iam::000000000000:role/pipe-role \ --source-parameters { SqsQueueParameters: { BatchSize: 10, DeadLetterConfig: {Arn: arn:aws:sqs:us-east-1:000000000000:dlq-queue} } } \ --endpoint-url $AWS_ENDPOINT_URL投遞失敗時(shí)sendToDeadLetterQueue 會(huì)把失敗 payload 發(fā)送到 DLQ若未配置 DLQ則返回投遞失敗計(jì)數(shù)并記錄告警日志。批次大小與輪詢行為默認(rèn)BatchSize為 10可通過(guò)各源參數(shù)塊的BatchSize字段調(diào)整getBatchSize輪詢間隔固定為 1 秒POLL_INTERVAL_MS 1000通過(guò) Vert.x 周期定時(shí)器驅(qū)動(dòng)并使用獨(dú)立線程池執(zhí)行保證同一管道不會(huì)并發(fā)輪詢activePolls防重入見 PipesPollerKinesis 迭代器過(guò)期ExpiredIteratorException或 DynamoDB 流數(shù)據(jù)被裁剪TrimmedDataAccessException時(shí)會(huì)自動(dòng)重置迭代器并重新拉取pollKinesis、pollDynamoDbStreamsKafka 路徑下REST Proxy 一次輪詢可能返回超過(guò)一個(gè)BatchSize的記錄因此一個(gè)輪詢周期會(huì)順序投遞多個(gè)批次一旦某分區(qū)投遞失敗該分區(qū)在后續(xù)批次中也不會(huì)提交偏移量保證“先失敗記錄重投”的順序語(yǔ)義deliverKafkaBatch。與 CloudFormation 的集成Pipes 服務(wù)實(shí)現(xiàn)了ResourceProvider接口支持 CloudFormation 對(duì)AWS::Pipes::Pipe資源的創(chuàng)建、更新與刪除提供器位于 cloudformation/provisioners 下的PipesCfnProvisioner并配套了回滾恢復(fù)測(cè)試 PipesCfnRollbackRestoresExactConfigurationTest.java 與 PipesCfnProvisionerTest.java。restorePipe路徑采用“null 即清空”的全量恢復(fù)語(yǔ)義確保 CloudFormation 回滾時(shí)能把管道精確還原到快照狀態(tài)PipesService.restorePipe。測(cè)試與驗(yàn)證倉(cāng)庫(kù)提供了多層測(cè)試覆蓋 pipes 服務(wù)單元測(cè)試PipesServiceTest.java、PipesPollerTest.java、PipesTargetInvokerTest.java、PipesFilterMatcherTest.java、KarapaceManagerTest.java、PipesKafkaRestClientTest.java集成測(cè)試PipesIntegrationTest.java、PipesPollerIntegrationTest.javaAWS CLI 兼容性測(cè)試pipes.bats 覆蓋 create/describe/list/update/start/stop 全流程SDK 各語(yǔ)言套件sdk-test-go、sdk-test-node、sdk-test-java、sdk-test-python中的pipes用例驗(yàn)證多語(yǔ)言 SDK 兼容性。結(jié)語(yǔ)Floci 的 EventBridge Pipes 實(shí)現(xiàn)覆蓋了從創(chuàng)建、啟停、更新、刪除到列表過(guò)濾的完整生命周期并對(duì) SQS/Kinesis/DynamoDB Streams/Kafka 四種源與 Lambda/SQS/SNS/Kinesis/Step Functions/EventBridge 多種目標(biāo)提供事件搬運(yùn)能力Kafka 源通過(guò)按需啟動(dòng)的 Karapace REST 邊車實(shí)現(xiàn)零內(nèi)嵌依賴的輪詢。在使用時(shí)請(qǐng)?zhí)貏e注意當(dāng)前仿真邊界ParallelizationFactor僅為線格式兼容單分片順序輪詢、enrichment 僅支持 Lambda 且僅作用于 SQS 源路徑、Kafka 源依賴 Docker。這些限制都以代碼注釋和文檔!!! note塊的形式明確標(biāo)注便于你在搭建本地事件管道時(shí)做出準(zhǔn)確預(yù)期。【免費(fèi)下載鏈接】flociLight, fluffy, and always free - The AWS Local Emulator alternative項(xiàng)目地址: https://gitcode.com/gh_mirrors/fl/floci創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考