BackupMessagesLoader { public void loadMessages(List<BackupMessage> messages) { logger.info("Loading {} messages from backup storage.", messages.size()); toResend.set(new ConcurrentLinkedQueue<>()); sendMessages(messages); if (toResend.get().size() == 0) { logger.info("No messages to resend."); return; } int retry = 0; do { if (retry > 0) { List<Pair<Message, CachedTopic>> retryMessages = Lists.newArrayList(toResend.getAndSet(new ConcurrentLinkedQueue<>())); resendMessages(retryMessages, retry); } try { Thread.sleep(resendSleep); } catch (InterruptedException e) { logger.warn("Sleep interrupted", e); } retry++; } while (toResend.get().size() > 0 && retry <= maxResendRetries); logger.info("Finished resending messages from backup storage after retry #{} with {} unsent messages.", retry - 1, toResend.get().size()); } @Inject  BackupMessagesLoader(BrokerMessageProducer brokerMessageProducer,
                                BrokerListeners brokerListeners,
                                TopicsCache topicsCache,
                                Trackers trackers,
                                ConfigFactory config); void loadMessages(List<BackupMessage> messages); void loadFromTemporaryBackupV2File(File file); void clearTopicsAvailabilityCache();  }
@Test public void shouldSendAndResendMessages() { int noOfSentCalls = 2; when(configFactory.getIntProperty(Configs.MESSAGES_LOCAL_STORAGE_MAX_AGE_HOURS)).thenReturn(8); when(configFactory.getIntProperty(Configs.MESSAGES_LOCAL_STORAGE_MAX_RESEND_RETRIES)).thenReturn(noOfSentCalls - 1); when(trackers.get(eq(topic))).thenReturn(new NoOperationPublishingTracker()); doAnswer(invocation -> { Object[] args = invocation.getArguments(); ((PublishingCallback) args[2]).onUnpublished((Message) args[0], ((CachedTopic) args[1]).getTopic(), new Exception("test")); return ""; }).when(producer).send(any(JsonMessage.class), eq(cachedTopic), any(PublishingCallback.class)); MessageRepository messageRepository = new ChronicleMapMessageRepository( new File(tempDir.getAbsoluteFile(), "messages.dat"), ENTRIES, AVERAGE_MESSAGE_SIZE ); BackupMessagesLoader backupMessagesLoader = new BackupMessagesLoader(producer, listeners, topicsCache, trackers, configFactory); messageRepository.save(messageOfAge(1), topic); backupMessagesLoader.loadMessages(messageRepository.findAll()); verify(producer, times(noOfSentCalls)).send(any(JsonMessage.class), eq(cachedTopic), any(PublishingCallback.class)); verify(listeners, times(noOfSentCalls)).onError(any(JsonMessage.class), eq(topic), any(Exception.class)); }
@Test public void shouldSendOnlyWhenBrokerTopicIsAvailable() { when(configFactory.getIntProperty(Configs.MESSAGES_LOCAL_STORAGE_MAX_AGE_HOURS)).thenReturn(10); when(producer.isTopicAvailable(cachedTopic)).thenReturn(false).thenReturn(false).thenReturn(true); BackupMessagesLoader backupMessagesLoader = new BackupMessagesLoader(producer, listeners, topicsCache, trackers, configFactory); MessageRepository messageRepository = new ChronicleMapMessageRepository( new File(tempDir.getAbsoluteFile(), "messages.dat"), ENTRIES, AVERAGE_MESSAGE_SIZE ); messageRepository.save(messageOfAge(1), topic); backupMessagesLoader.loadMessages(messageRepository.findAll()); verify(producer, times(3)).isTopicAvailable(cachedTopic); verify(producer, times(1)).send(any(JsonMessage.class), eq(cachedTopic), any(PublishingCallback.class)); }
@Test public void shouldNotSendOldMessages() { when(configFactory.getIntProperty(Configs.MESSAGES_LOCAL_STORAGE_MAX_AGE_HOURS)).thenReturn(8); when(configFactory.getIntProperty(Configs.MESSAGES_LOCAL_STORAGE_MAX_RESEND_RETRIES)).thenReturn(2); MessageRepository messageRepository = new ChronicleMapMessageRepository( new File(tempDir.getAbsoluteFile(), "messages.dat"), ENTRIES, AVERAGE_MESSAGE_SIZE ); BackupMessagesLoader backupMessagesLoader = new BackupMessagesLoader(producer, listeners, topicsCache, trackers, configFactory); messageRepository.save(messageOfAge(1), topic); messageRepository.save(messageOfAge(10), topic); messageRepository.save(messageOfAge(10), topic); backupMessagesLoader.loadMessages(messageRepository.findAll()); verify(producer, times(1)).send(any(JsonMessage.class), eq(cachedTopic), any(PublishingCallback.class)); }