public FailoverResponse failover() throws Throwable {
            LOGGER.info("failover start {}", request);
            int nodeId = request.getNodeId();
            long nodeEpoch = request.getNodeEpoch();

            FailoverResponse resp = new FailoverResponse();
            resp.setNodeId(nodeId);
            resp.setEpoch(nodeEpoch);

            // fence the device to ensure the old node stops writing to the delta WAL
            // recover WAL data and upload to S3
            WriteAheadLog wal = factory.getWal(request);
            try {
                wal.start();
            } catch (WALNotInitializedException ex) {
                LOGGER.info("fail over empty wal {}", request);
                return resp;
            }

            try {
                WALMetadata metadata = wal.metadata();
                if (nodeId != metadata.nodeId()) {
                    throw new IllegalArgumentException(String.format("nodeId mismatch, request=%s, wal=%s", request, metadata));
                }
                if (nodeEpoch < metadata.epoch()) {
                    throw new IllegalStateException(String.format("epoch mismatch, request=%s, wal=%s", request, metadata));
                }

                Logger taskLogger = new LogContext(String.format("[Failover nodeId=%s epoch=%s]", nodeId, nodeEpoch)).logger(FailoverTask.class);
                StreamManager streamManager = factory.getStreamManager(nodeId, nodeEpoch);
                ObjectManager objectManager = factory.getObjectManager(nodeId, nodeEpoch);
                LOGGER.info("failover recover {}", request);
                walRecover.recover(wal, streamManager, objectManager, taskLogger);
            } finally {
                wal.shutdownGracefully();
            }
            LOGGER.info("failover done {}", request);
            return resp;
        }
To perform a Delta WAL failover, follow these steps:
1. Ensure the old node stops writing to the delta WAL.
2. Instruct the controller to reject all requests sent by the old node.
3. Upload the delta WAL to S3.
4. Lastly, close any streams that were opened by the old node and are currently active.
s3stream/src/main/java/com/automq/stream/s3/failover/Failover.java