「布道师系列文章」小红书黄章衡：AutoMQ Serverless 基石 - 秒级分区迁移

ReplicationControlManager：changePartitionReassignmentV2 {
    PartitionChangeBuilder builder = new PartitionChangeBuilder(part,        tp.topicId(),        tp.partitionId(),        // no leader election, isAcceptableLeader 直接返回 False，代表不选主        brokerId -> false,        featureControl.metadataVersion(),        getTopicEffectiveMinIsr(topics.get(tp.topicId()).name.toString())    );    builder.setZkMigrationEnabled(clusterControl.zkRegistrationAllowed());    builder.setEligibleLeaderReplicasEnabled(isElrEnabled());    // 设置 ISR、Replicas 为 [target.replicas().get(0)]    builder.setTargetNode(target.replicas().get(0));    TopicControlInfo topicControlInfo = topics.get(tp.topicId());    if (topicControlInfo == null) {        log.warn("unknown topicId[{}]", tp.topicId());    } else {        // 选主超时器        TopicPartition topicPartition = new TopicPartition(topicControlInfo.name, tp.partitionId());        addPartitionToReElectTimeouts(topicPartition);    }    return builder.setDefaultDirProvider(clusterDescriber).build();}

ElasticReplicaManager: asyncApplyDelta(delta: TopicsDelta, newImage: MetadataImage) {    if (!localChanges.deletes.isEmpty) {      val deletes = localChanges.deletes.asScala        .map { tp =>          val isCurrentLeader = Option(delta.image().getTopic(tp.topic()))            .map(image => image.partitions().get(tp.partition()))            .exists(partition => partition.leader == config.nodeId)          val deleteRemoteLog = delta.topicWasDeleted(tp.topic()) && isCurrentLeader          StopPartition(tp, deleteLocalLog = true, deleteRemoteLog = deleteRemoteLog)        }        .toSet          def doPartitionDeletion(): Unit = {        stateChangeLogger.info(s"Deleting ${deletes.size} partition(s).")        deletes.foreach(stopPartition => {          val opCf = doPartitionDeletionAsyncLocked(stopPartition)          opCfList.add(opCf)        })      }          doPartitionDeletion()    }    }

ElasticLog close(): CompletableFuture[Void] = {    // already flush in UnifiedLog#close, so it's safe to set cleaned shutdown.    // 标志为 Clean Shutdown    partitionMeta.setCleanedShutdown(true)    partitionMeta.setStartOffset(logStartOffset)    partitionMeta.setRecoverOffset(recoveryPoint)
    maybeHandleIOException(s"Error while closing $topicPartition in dir ${dir.getParent}") {        // 持久化元数据        CoreUtils.swallow(persistLogMeta(), this)        CoreUtils.swallow(checkIfMemoryMappedBufferClosed(), this)        CoreUtils.swallow(segments.close(), this)        CoreUtils.swallow(persistPartitionMeta(), this)    }    info("log(except for streams) closed")    // 关闭分区对应的所有 Streams    closeStreams()}

S3Stream：Close（）{
    // await all pending append/fetch/trim request    List<CompletableFuture<?>> pendingRequests = new ArrayList<>(pendingAppends);    if (GlobalSwitch.STRICT) {        pendingRequests.addAll(pendingFetches);    }    pendingRequests.add(lastPendingTrim);    CompletableFuture<Void> awaitPendingRequestsCf = CompletableFuture.allOf(pendingRequests.toArray(new CompletableFuture[0]));    CompletableFuture<Void> closeCf = new CompletableFuture<>();        // Close0 函数触发强制上传和 Stream 关闭    awaitPendingRequestsCf.whenComplete((nil, ex) -> propagate(exec(this::close0, LOGGER, "close"), closeCf));    }
private CompletableFuture<Void> close0() {    return storage.forceUpload(streamId)        .thenCompose(nil -> streamManager.closeStream(streamId, epoch));}

ElasticReplicaManager:StopPartitions(partitionsToStop: collection.Set[StopPartition]) {
    partitionsToStop.foreach { stopPartition =>      val topicPartition = stopPartition.topicPartition      if (stopPartition.deleteLocalLog) {        getPartition(topicPartition) match {          case hostedPartition: HostedPartition.Online =>            if (allPartitions.remove(topicPartition, hostedPartition)) {              maybeRemoveTopicMetrics(topicPartition.topic)              // AutoMQ for Kafka inject start              if (ElasticLogManager.enabled()) {                // For elastic stream, partition leader alter is triggered by setting isr/replicas.                // When broker is not response for the partition, we need to close the partition                // instead of delete the partition.                val start = System.currentTimeMillis()                hostedPartition.partition.close().get()                info(s"partition $topicPartition is closed, cost ${System.currentTimeMillis() - start} ms, trigger leader election")                                // 主动触发选主                alterPartitionManager.tryElectLeader(topicPartition)              } else {                // Logs are not deleted here. They are deleted in a single batch later on.                // This is done to avoid having to checkpoint for every deletions.                hostedPartition.partition.delete()              }              // AutoMQ for Kafka inject end            }              case _ =>        }        partitionsToDelete += topicPartition      }
}

metaStream = if (metaNotExists) {    val stream = createMetaStream(client, key, replicationFactor, leaderEpoch, logIdent = logIdent)    info(s"${logIdent}created a new meta stream: stream_id=${stream.streamId()}")    stream} else {    val metaStreamId = Unpooled.wrappedBuffer(value.get()).readLong()    // open partition meta stream    val stream = client.streamClient().openStream(metaStreamId, OpenStreamOptions.builder().epoch(leaderEpoch).build())        .thenApply(stream => new MetaStream(stream, META_SCHEDULE_EXECUTOR, logIdent))        .get()    info(s"${logIdent}opened existing meta stream: stream_id=$metaStreamId")    stream}

// load meta info for this partitionval partitionMetaOpt = metaMap.get(MetaStream.PARTITION_META_KEY).map(m => m.asInstanceOf[ElasticPartitionMeta])if (partitionMetaOpt.isEmpty) {    partitionMeta = new ElasticPartitionMeta(0, 0, 0)    persistMeta(metaStream, MetaKeyValue.of(MetaStream.PARTITION_META_KEY, ElasticPartitionMeta.encode(partitionMeta)))} else {    partitionMeta = partitionMetaOpt.get}info(s"${logIdent}loaded partition meta: $partitionMeta")
//load producer snapshots for this partitionval producerSnapshotsMeta = metaMap.get(MetaStream.PRODUCER_SNAPSHOTS_META_KEY).map(m => m.asInstanceOf[ElasticPartitionProducerSnapshotsMeta]).getOrElse(new ElasticPartitionProducerSnapshotsMeta())val snapshotsMap = new ConcurrentSkipListMap[java.lang.Long, ByteBuffer](producerSnapshotsMeta.getSnapshots)if (!snapshotsMap.isEmpty) {    info(s"${logIdent}loaded ${snapshotsMap.size} producer snapshots, offsets(filenames) are ${snapshotsMap.keySet()} ")} else {    info(s"${logIdent}loaded no producer snapshots")}
// load leader epoch checkpointval leaderEpochCheckpointMetaOpt = metaMap.get(MetaStream.LEADER_EPOCH_CHECKPOINT_KEY).map(m => m.asInstanceOf[ElasticLeaderEpochCheckpointMeta])val leaderEpochCheckpointMeta = if (leaderEpochCheckpointMetaOpt.isEmpty) {    val newMeta = new ElasticLeaderEpochCheckpointMeta(LeaderEpochCheckpointFile.CURRENT_VERSION, List.empty[EpochEntry].asJava)    // save right now.    persistMeta(metaStream, MetaKeyValue.of(MetaStream.LEADER_EPOCH_CHECKPOINT_KEY, ByteBuffer.wrap(newMeta.encode())))    newMeta} else {    leaderEpochCheckpointMetaOpt.get}info(s"${logIdent}loaded leader epoch checkpoint with ${leaderEpochCheckpointMeta.entries.size} entries")if (!leaderEpochCheckpointMeta.entries.isEmpty) {    val lastEntry = leaderEpochCheckpointMeta.entries.get(leaderEpochCheckpointMeta.entries.size - 1)    info(s"${logIdent}last leaderEpoch entry is: $lastEntry")}

val logMeta: ElasticLogMeta = metaMap.get(MetaStream.LOG_META_KEY).map(m => m.asInstanceOf[ElasticLogMeta]).getOrElse(new ElasticLogMeta())logStreamManager = new ElasticLogStreamManager(logMeta.getStreamMap, client.streamClient(), replicationFactor, leaderEpoch)val streamSliceManager = new ElasticStreamSliceManager(logStreamManager)
val logSegmentManager = new ElasticLogSegmentManager(metaStream, logStreamManager, logIdent)
// load LogSegments and recover logval segments = new CachedLogSegments(topicPartition)// 这里通过 ElasticLogLoader 恢复所有 elastic log segment 的状态val offsets = new ElasticLogLoader(    logMeta,    segments,    logSegmentManager,    streamSliceManager,    dir,    topicPartition,    config,    time,    hadCleanShutdown = partitionMeta.getCleanedShutdown,    logStartOffsetCheckpoint = partitionMeta.getStartOffset,    partitionMeta.getRecoverOffset,    Optional.empty(),    producerStateManager = producerStateManager,    numRemainingSegments = numRemainingSegments,    createAndSaveSegmentFunc = createAndSaveSegment(logSegmentManager, logIdent = logIdent)).load()info(s"${logIdent}loaded log meta: $logMeta")