06-事务与Exactly-Once语义

// 幂等Producer结构
type IdempotentProducer struct {
    producerID    int64
    epoch         int16
    sequenceNumber int32
    brokerClients map[int32]*BrokerClient
    stateManager  *ProducerStateManager
}

// Producer状态管理器
type ProducerStateManager struct {
    producerID    int64
    epoch         int16
    sequenceNumber int32
    lastSequence  map[string]int32 // Topic-Partition -> Last Sequence
    mu            sync.RWMutex
}

// 发送消息
func (ip *IdempotentProducer) SendMessage(topic string, partition int32, key, value []byte) error {
    // 1. 生成序列号
    sequenceNumber := ip.getNextSequenceNumber(topic, partition)
    
    // 2. 构建消息
    message := &ProducerRecord{
        Topic:     topic,
        Partition: partition,
        Key:       key,
        Value:     value,
        ProducerID: ip.producerID,
        Epoch:     ip.epoch,
        SequenceNumber: sequenceNumber,
    }
    
    // 3. 发送到Broker
    return ip.sendToBroker(message)
}

// 获取下一个序列号
func (ip *IdempotentProducer) getNextSequenceNumber(topic string, partition int32) int32 {
    ip.stateManager.mu.Lock()
    defer ip.stateManager.mu.Unlock()
    
    key := fmt.Sprintf("%s-%d", topic, partition)
    lastSeq := ip.stateManager.lastSequence[key]
    nextSeq := lastSeq + 1
    
    // 更新序列号
    ip.stateManager.lastSequence[key] = nextSeq
    
    return nextSeq
}

// 发送到Broker
func (ip *IdempotentProducer) sendToBroker(message *ProducerRecord) error {
    // 获取Broker客户端
    brokerClient := ip.brokerClients[message.Partition]
    if brokerClient == nil {
        return fmt.Errorf("无法找到Broker客户端")
    }
    
    // 构建请求
    request := &ProduceRequest{
        Topic:     message.Topic,
        Partition: message.Partition,
        Records:   []*ProducerRecord{message},
        ProducerID: message.ProducerID,
        Epoch:     message.Epoch,
        SequenceNumber: message.SequenceNumber,
    }
    
    // 发送请求
    response, err := brokerClient.Produce(request)
    if err != nil {
        return err
    }
    
    // 处理响应
    return ip.handleProduceResponse(response)
}

// Broker端幂等处理
type BrokerIdempotentHandler struct {
    producerStateManager *BrokerProducerStateManager
}

// 处理生产请求
func (bih *BrokerIdempotentHandler) handleProduceRequest(request *ProduceRequest) (*ProduceResponse, error) {
    // 1. 验证Producer ID和Epoch
    if err := bih.validateProducerID(request.ProducerID, request.Epoch); err != nil {
        return nil, err
    }
    
    // 2. 检查序列号
    if err := bih.validateSequenceNumber(request); err != nil {
        return nil, err
    }
    
    // 3. 处理消息
    return bih.processMessages(request)
}

// 验证Producer ID
func (bih *BrokerIdempotentHandler) validateProducerID(producerID int64, epoch int16) error {
    state := bih.producerStateManager.getProducerState(producerID)
    
    if state == nil {
        // 新Producer，记录状态
        bih.producerStateManager.addProducerState(producerID, epoch)
        return nil
    }
    
    // 检查Epoch
    if epoch < state.Epoch {
        return fmt.Errorf("过期的Producer Epoch: %d < %d", epoch, state.Epoch)
    }
    
    if epoch > state.Epoch {
        // 更新Epoch
        state.Epoch = epoch
        state.SequenceNumber = -1 // 重置序列号
    }
    
    return nil
}

// 验证序列号
func (bih *BrokerIdempotentHandler) validateSequenceNumber(request *ProduceRequest) error {
    producerID := request.ProducerID
    epoch := request.Epoch
    sequenceNumber := request.SequenceNumber
    
    state := bih.producerStateManager.getProducerState(producerID)
    if state == nil {
        return fmt.Errorf("Producer状态不存在")
    }
    
    // 检查序列号连续性
    expectedSequence := state.SequenceNumber + 1
    if sequenceNumber < expectedSequence {
        // 重复消息，直接返回成功
        return nil
    }
    
    if sequenceNumber > expectedSequence {
        return fmt.Errorf("序列号不连续: %d > %d", sequenceNumber, expectedSequence)
    }
    
    // 更新序列号
    state.SequenceNumber = sequenceNumber
    
    return nil
}

// 事务管理器
type TransactionManager struct {
    transactionID    string
    producerID       int64
    epoch            int16
    coordinator      *TransactionCoordinator
    state            TransactionState
    partitions       map[string][]int32 // Topic -> Partitions
    offsets          map[string]map[int32]int64 // Topic -> Partition -> Offset
    startTime        int64
    timeoutMs        int64
}

// 事务状态
type TransactionState int

const (
    Empty TransactionState = iota
    Ongoing
    PrepareCommit
    PrepareAbort
    CompleteCommit
    CompleteAbort
    Dead
)

// 开始事务
func (tm *TransactionManager) beginTransaction() error {
    // 1. 生成事务ID
    tm.transactionID = tm.generateTransactionID()
    
    // 2. 设置状态
    tm.state = Ongoing
    tm.startTime = time.Now().UnixMilli()
    
    // 3. 初始化分区和偏移量
    tm.partitions = make(map[string][]int32)
    tm.offsets = make(map[string]map[int32]int64)
    
    // 4. 通知Coordinator
    return tm.coordinator.beginTransaction(tm.transactionID, tm.producerID, tm.epoch)
}

// 发送消息
func (tm *TransactionManager) sendMessage(topic string, partition int32, key, value []byte) error {
    // 1. 验证事务状态
    if tm.state != Ongoing {
        return fmt.Errorf("事务状态不正确: %v", tm.state)
    }
    
    // 2. 记录分区
    if tm.partitions[topic] == nil {
        tm.partitions[topic] = make([]int32, 0)
    }
    tm.partitions[topic] = append(tm.partitions[topic], partition)
    
    // 3. 发送消息
    message := &ProducerRecord{
        Topic:     topic,
        Partition: partition,
        Key:       key,
        Value:     value,
        ProducerID: tm.producerID,
        Epoch:     tm.epoch,
        TransactionID: tm.transactionID,
    }
    
    return tm.sendToBroker(message)
}

// 提交事务
func (tm *TransactionManager) commitTransaction() error {
    // 1. 设置状态
    tm.state = PrepareCommit
    
    // 2. 通知Coordinator
    if err := tm.coordinator.prepareCommit(tm.transactionID); err != nil {
        return err
    }
    
    // 3. 等待协调结果
    if err := tm.waitForCoordinatorDecision(); err != nil {
        return err
    }
    
    // 4. 执行提交
    if tm.state == CompleteCommit {
        return tm.executeCommit()
    } else {
        return tm.executeAbort()
    }
}

// 执行提交
func (tm *TransactionManager) executeCommit() error {
    // 1. 提交所有消息
    for topic, partitions := range tm.partitions {
        for _, partition := range partitions {
            if err := tm.commitPartition(topic, partition); err != nil {
                return err
            }
        }
    }
    
    // 2. 提交偏移量
    for topic, partitionOffsets := range tm.offsets {
        for partition, offset := range partitionOffsets {
            if err := tm.commitOffset(topic, partition, offset); err != nil {
                return err
            }
        }
    }
    
    // 3. 更新状态
    tm.state = CompleteCommit
    
    return nil
}

// 事务协调器
type TransactionCoordinator struct {
    brokerID        int32
    transactionLog  *TransactionLog
    transactionState map[string]*TransactionState
    timeoutManager  *TransactionTimeoutManager
    mu              sync.RWMutex
}

// 事务状态
type TransactionState struct {
    transactionID    string
    producerID       int64
    epoch            int16
    state            TransactionState
    partitions       map[string][]int32
    offsets          map[string]map[int32]int64
    startTime        int64
    timeoutMs        int64
    lastUpdateTime   int64
}

// 开始事务
func (tc *TransactionCoordinator) beginTransaction(transactionID string, producerID int64, epoch int16) error {
    tc.mu.Lock()
    defer tc.mu.Unlock()
    
    // 1. 创建事务状态
    state := &TransactionState{
        transactionID:  transactionID,
        producerID:     producerID,
        epoch:          epoch,
        state:          Ongoing,
        partitions:     make(map[string][]int32),
        offsets:        make(map[string]map[int32]int64),
        startTime:      time.Now().UnixMilli(),
        lastUpdateTime: time.Now().UnixMilli(),
    }
    
    // 2. 存储状态
    tc.transactionState[transactionID] = state
    
    // 3. 写入事务日志
    if err := tc.transactionLog.writeTransactionState(state); err != nil {
        return err
    }
    
    // 4. 设置超时
    tc.timeoutManager.setTimeout(transactionID, state.timeoutMs)
    
    return nil
}

// 准备提交
func (tc *TransactionCoordinator) prepareCommit(transactionID string) error {
    tc.mu.Lock()
    defer tc.mu.Unlock()
    
    // 1. 获取事务状态
    state, exists := tc.transactionState[transactionID]
    if !exists {
        return fmt.Errorf("事务不存在: %s", transactionID)
    }
    
    // 2. 验证状态
    if state.state != Ongoing {
        return fmt.Errorf("事务状态不正确: %v", state.state)
    }
    
    // 3. 更新状态
    state.state = PrepareCommit
    state.lastUpdateTime = time.Now().UnixMilli()
    
    // 4. 写入事务日志
    if err := tc.transactionLog.writeTransactionState(state); err != nil {
        return err
    }
    
    // 5. 通知所有相关Broker
    return tc.notifyBrokersPrepareCommit(state)
}

// 通知Broker准备提交
func (tc *TransactionCoordinator) notifyBrokersPrepareCommit(state *TransactionState) error {
    // 获取所有相关的Broker
    brokers := tc.getRelatedBrokers(state)
    
    // 并行通知所有Broker
    var wg sync.WaitGroup
    var errors []error
    var mu sync.Mutex
    
    for _, brokerID := range brokers {
        wg.Add(1)
        go func(bID int32) {
            defer wg.Done()
            
            client := tc.getBrokerClient(bID)
            request := &PrepareCommitRequest{
                TransactionID: state.transactionID,
                ProducerID:    state.producerID,
                Epoch:         state.epoch,
                Partitions:    state.partitions,
            }
            
            if err := client.PrepareCommit(request); err != nil {
                mu.Lock()
                errors = append(errors, err)
                mu.Unlock()
            }
        }(brokerID)
    }
    
    wg.Wait()
    
    // 检查是否有错误
    if len(errors) > 0 {
        return fmt.Errorf("通知Broker失败: %v", errors)
    }
    
    return nil
}

// 完成提交
func (tc *TransactionCoordinator) completeCommit(transactionID string) error {
    tc.mu.Lock()
    defer tc.mu.Unlock()
    
    // 1. 获取事务状态
    state, exists := tc.transactionState[transactionID]
    if !exists {
        return fmt.Errorf("事务不存在: %s", transactionID)
    }
    
    // 2. 更新状态
    state.state = CompleteCommit
    state.lastUpdateTime = time.Now().UnixMilli()
    
    // 3. 写入事务日志
    if err := tc.transactionLog.writeTransactionState(state); err != nil {
        return err
    }
    
    // 4. 通知所有相关Broker
    return tc.notifyBrokersCompleteCommit(state)
}

// 事务超时处理
func (tc *TransactionCoordinator) handleTransactionTimeout(transactionID string) error {
    tc.mu.Lock()
    defer tc.mu.Unlock()
    
    // 1. 获取事务状态
    state, exists := tc.transactionState[transactionID]
    if !exists {
        return nil // 事务已不存在
    }
    
    // 2. 检查是否超时
    if time.Now().UnixMilli()-state.startTime < state.timeoutMs {
        return nil // 未超时
    }
    
    // 3. 执行回滚
    state.state = CompleteAbort
    state.lastUpdateTime = time.Now().UnixMilli()
    
    // 4. 写入事务日志
    if err := tc.transactionLog.writeTransactionState(state); err != nil {
        return err
    }
    
    // 5. 通知所有相关Broker
    return tc.notifyBrokersAbort(state)
}

// 读隔离级别
type IsolationLevel int

const (
    ReadUncommitted IsolationLevel = iota // 读取未提交
    ReadCommitted                         // 读取已提交
)

// 隔离级别管理器
type IsolationLevelManager struct {
    isolationLevel IsolationLevel
    hwmManager     *HighWatermarkManager
}

// 过滤消息
func (ilm *IsolationLevelManager) filterMessages(records []Record, isolationLevel IsolationLevel) []Record {
    switch isolationLevel {
    case ReadUncommitted:
        return records // 返回所有消息
    case ReadCommitted:
        return ilm.filterCommittedMessages(records)
    default:
        return records
    }
}

// 过滤已提交的消息
func (ilm *IsolationLevelManager) filterCommittedMessages(records []Record) []Record {
    var committedRecords []Record
    
    for _, record := range records {
        // 检查消息是否已提交
        if ilm.isMessageCommitted(record) {
            committedRecords = append(committedRecords, record)
        }
    }
    
    return committedRecords
}

// 检查消息是否已提交
func (ilm *IsolationLevelManager) isMessageCommitted(record Record) bool {
    // 1. 检查是否为事务消息
    if record.TransactionID == "" {
        return true // 非事务消息，已提交
    }
    
    // 2. 检查事务状态
    transactionState := ilm.getTransactionState(record.TransactionID)
    if transactionState == nil {
        return false // 事务不存在，未提交
    }
    
    // 3. 检查事务是否已提交
    return transactionState.state == CompleteCommit
}

// 获取事务状态
func (ilm *IsolationLevelManager) getTransactionState(transactionID string) *TransactionState {
    // 从事务日志获取状态
    return ilm.transactionLog.getTransactionState(transactionID)
}

// 消费者隔离级别处理
type ConsumerIsolationHandler struct {
    isolationLevel IsolationLevel
    isolationManager *IsolationLevelManager
}

// 处理拉取请求
func (cih *ConsumerIsolationHandler) handleFetchRequest(request *FetchRequest) (*FetchResponse, error) {
    // 1. 从日志读取消息
    records, err := cih.readFromLog(request.Topic, request.Partition, request.Offset, request.MaxBytes)
    if err != nil {
        return nil, err
    }
    
    // 2. 根据隔离级别过滤消息
    filteredRecords := cih.isolationManager.filterMessages(records, cih.isolationLevel)
    
    // 3. 返回响应
    return &FetchResponse{
        Topic:     request.Topic,
        Partition: request.Partition,
        Records:   filteredRecords,
        HighWatermark: cih.getHighWatermark(request.Topic, request.Partition),
    }, nil
}

// 读取日志
func (cih *ConsumerIsolationHandler) readFromLog(topic string, partition int32, offset int64, maxBytes int64) ([]Record, error) {
    // 从分区日志读取消息
    return cih.partitionLog.Fetch(topic, partition, offset, maxBytes)
}

// 获取高水位
func (cih *ConsumerIsolationHandler) getHighWatermark(topic string, partition int32) int64 {
    return cih.hwmManager.getHighWatermark(topic, partition)
}

// EOS管理器
type EOSManager struct {
    producerManager *ProducerManager
    consumerManager *ConsumerManager
    transactionManager *TransactionManager
    isolationManager *IsolationLevelManager
}

// 端到端EOS保证
func (eos *EOSManager) ensureEOS() error {
    // 1. 配置幂等Producer
    if err := eos.configureIdempotentProducer(); err != nil {
        return err
    }
    
    // 2. 配置事务
    if err := eos.configureTransaction(); err != nil {
        return err
    }
    
    // 3. 配置读隔离级别
    if err := eos.configureIsolationLevel(); err != nil {
        return err
    }
    
    // 4. 配置消费者事务
    if err := eos.configureConsumerTransaction(); err != nil {
        return err
    }
    
    return nil
}

// 配置幂等Producer
func (eos *EOSManager) configureIdempotentProducer() error {
    // 启用幂等性
    eos.producerManager.enableIdempotence = true
    
    // 设置重试次数
    eos.producerManager.retries = 3
    
    // 设置最大飞行请求数
    eos.producerManager.maxInFlightRequests = 1
    
    return nil
}

// 配置事务
func (eos *EOSManager) configureTransaction() error {
    // 启用事务
    eos.transactionManager.enabled = true
    
    // 设置事务超时
    eos.transactionManager.timeoutMs = 30000 // 30秒
    
    // 设置事务ID
    eos.transactionManager.transactionID = eos.generateTransactionID()
    
    return nil
}

// 配置读隔离级别
func (eos *EOSManager) configureIsolationLevel() error {
    // 设置为读已提交
    eos.isolationManager.isolationLevel = ReadCommitted
    
    return nil
}

// 配置消费者事务
func (eos *EOSManager) configureConsumerTransaction() error {
    // 启用事务消费
    eos.consumerManager.enableAutoCommit = false
    
    // 设置读隔离级别
    eos.consumerManager.isolationLevel = ReadCommitted
    
    return nil
}

// 事务性消费
func (eos *EOSManager) transactionalConsume() error {
    // 1. 开始事务
    if err := eos.transactionManager.beginTransaction(); err != nil {
        return err
    }
    
    // 2. 消费消息
    messages, err := eos.consumerManager.poll()
    if err != nil {
        return err
    }
    
    // 3. 处理消息
    for _, message := range messages {
        if err := eos.processMessage(message); err != nil {
            // 处理失败，回滚事务
            eos.transactionManager.abortTransaction()
            return err
        }
    }
    
    // 4. 提交事务
    if err := eos.transactionManager.commitTransaction(); err != nil {
        return err
    }
    
    return nil
}

// 处理消息
func (eos *EOSManager) processMessage(message *ConsumerRecord) error {
    // 1. 业务逻辑处理
    if err := eos.businessLogic.process(message); err != nil {
        return err
    }
    
    // 2. 记录处理结果
    if err := eos.recordProcessingResult(message); err != nil {
        return err
    }
    
    return nil
}

// 性能优化配置
type PerformanceOptimizer struct {
    batchSize          int
    lingerMs           int
    compressionType    string
    maxInFlightRequests int
    transactionTimeout int64
}

// 优化配置
func (po *PerformanceOptimizer) optimizeForEOS() *PerformanceOptimizer {
    return &PerformanceOptimizer{
        batchSize:          16384,  // 16KB批次
        lingerMs:           5,      // 5ms延迟
        compressionType:    "snappy", // Snappy压缩
        maxInFlightRequests: 1,     // 单飞行请求
        transactionTimeout: 30000,  // 30秒超时
    }
}

// 批量事务处理
func (po *PerformanceOptimizer) batchTransactionProcessing() error {
    // 1. 收集多个操作
    operations := po.collectOperations()
    
    // 2. 批量开始事务
    if err := po.batchBeginTransaction(operations); err != nil {
        return err
    }
    
    // 3. 批量执行操作
    if err := po.batchExecuteOperations(operations); err != nil {
        return err
    }
    
    // 4. 批量提交事务
    if err := po.batchCommitTransaction(operations); err != nil {
        return err
    }
    
    return nil
}

// 异步事务状态检查
func (po *PerformanceOptimizer) asyncTransactionStateCheck() {
    // 1. 异步检查事务状态
    go func() {
        for {
            po.checkTransactionStates()
            time.Sleep(100 * time.Millisecond)
        }
    }()
}

// 缓存事务状态
func (po *PerformanceOptimizer) cacheTransactionStates() {
    // 1. 缓存常用事务状态
    po.transactionStateCache = make(map[string]*TransactionState)
    
    // 2. 定期更新缓存
    go func() {
        for {
            po.updateTransactionStateCache()
            time.Sleep(1000 * time.Millisecond)
        }
    }()
}