Aeron档案 - 扩展现有记录

Question

我有一个Aeron档案馆，并希望扩展现有的录制，即它在服务重新启动后继续附加消息。

我找不到任何实际示例如何做到这一点，所以我根据Aeron Javadoc/ Aeron食谱。

到目前为止，我的目标

我正试图从档案中获得lastrecordingid，位置等。

private void findLatestRecording() {
        final RecordingDescriptorConsumer consumer =
                (controlSessionId, correlationId, recordingId,
                        startTimestamp, stopTimestamp, startPosition,
                        stopPosition, initialTermId, segmentFileLength,
                        termBufferLength, mtuLength, sessionId,
                        streamId, strippedChannel, originalChannel,
                        sourceIdentity) ->  {
                    AeronArchiveJournal.this.lastRecordingId = recordingId;
                    AeronArchiveJournal.this.lastRecordingPosition = stopPosition;
                    AeronArchiveJournal.this.initialTermId = initialTermId;
                    AeronArchiveJournal.this.termBufferLength = termBufferLength;
                };

        final long fromRecordingId = 0L;
        final int recordCount = 100;
        final int foundCount = archive.listRecordingsForUri(fromRecordingId, recordCount, AeronChannels.ipc(), AeronStreams.STREAM_ID_JOURNAL, consumer);

        if (foundCount == 0) {
            LOG.info("No previous recording found, will start a new one");
        }
    }

然后，我试图扩展录制，

 private void extendExistingRecording() {
        publication = aeron.addExclusivePublication(AeronChannels.ipc(), AeronStreams.STREAM_ID_JOURNAL);

        String channelUri = new ChannelUriStringBuilder()
                .media(CommonContext.IPC_MEDIA)
                .initialPosition(lastRecordingPosition, initialTermId, termBufferLength)
                .build();

        LOG.info("Extending existing recording");
        LOG.info("Recording id: {}", lastRecordingId);
        LOG.info("Channel URI: {}", channelUri);


        archive.extendRecording(lastRecordingId, channelUri, AeronStreams.STREAM_ID_JOURNAL, SourceLocation.LOCAL);

        LOG.info("Waiting for recording to start for session {}", publication.sessionId());
        final CountersReader counters = aeron.countersReader();
        int counterId = RecordingPos.findCounterIdBySession(counters, publication.sessionId());

        while (CountersReader.NULL_COUNTER_ID == counterId) {
            journalIdleStrategy.idle();
            counterId = RecordingPos.findCounterIdBySession(counters, publication.sessionId());
        }
        lastRecordingId = RecordingPos.getRecordingId(counters, counterId);

    }

但是countersreader循环在警告后永远不会完成和Aeron打印：

io.aeron.archive.client.ArchiveEvent: WARN - cannot extend recording 0 image.joinPosition=0 != rec.stopPosition=25312

显然我缺少某些内容，但是到目前为止，我无法弄清楚什么。

Answer 1

为了扩展现有记录，有必要设置您将使用的出版物的初始位置来扩展现有记录。需要初始化该流，因此它是真正的扩展。这测试显示了如何完成。

https://github.com/real-logic/aeron/blob/master/aeron-system-tests/src/src/test/java/io/aeron/aeron/aeron/archive/extendrecordingtest.java