@Override
protected void masterOperation(final UpgradeSettingsRequest request, final ClusterState state, final ActionListener<AcknowledgedResponse> listener) {
    UpgradeSettingsClusterStateUpdateRequest clusterStateUpdateRequest = new UpgradeSettingsClusterStateUpdateRequest()
            .ackTimeout(request.timeout())
            .versions(request.versions())
            .masterNodeTimeout(request.masterNodeTimeout());

    updateSettingsService.upgradeIndexSettings(clusterStateUpdateRequest, new ActionListener<ClusterStateUpdateResponse>() {
        @Override
        public void onResponse(ClusterStateUpdateResponse response) {
            listener.onResponse(new AcknowledgedResponse(response.isAcknowledged()));
        }

        @Override
        public void onFailure(Exception t) {
            logger.debug(() -> new ParameterizedMessage("failed to upgrade minimum compatibility version settings on indices [{}]", request.versions().keySet()), t);
            listener.onFailure(t);
        }
    });
}
 

private void deleteIndexIteration(String[] toDelete) {
    for (String index : toDelete) {
        DeleteIndexRequest singleDeleteRequest = new DeleteIndexRequest(index);
        adminClient.indices().delete(singleDeleteRequest, ActionListener.wrap(singleDeleteResponse -> {
            if (!singleDeleteResponse.isAcknowledged()) {
                logger.error("Retrying deleting {} does not succeed.", index);
            }
        }, exception -> {
            if (exception instanceof IndexNotFoundException) {
                logger.info("{} was already deleted.", index);
            } else {
                logger.error(new ParameterizedMessage("Retrying deleting {} does not succeed.", index), exception);
            }
        }));
    }
}
 

protected EntryMessage entryMsg(final String format, final Object... params) {
    final int count = params == null ? 0 : params.length;
    if (count == 0) {
        if (Strings.isEmpty(format)) {
            return flowMessageFactory.newEntryMessage(null);
        }
        return flowMessageFactory.newEntryMessage(new SimpleMessage(format));
    }
    if (format != null) {
        return flowMessageFactory.newEntryMessage(new ParameterizedMessage(format, params));
    }
    final StringBuilder sb = new StringBuilder();
    sb.append("params(");
    for (int i = 0; i < count; i++) {
        if (i > 0) {
            sb.append(", ");
        }
        final Object parm = params[i];
        sb.append(parm instanceof Message ? ((Message) parm).getFormattedMessage() : String.valueOf(parm));
    }
    sb.append(')');
    return flowMessageFactory.newEntryMessage(new SimpleMessage(sb));
}
 

/**
 * Validates the given settings by running it through all update listeners without applying it. This
 * method will not change any settings but will fail if any of the settings can't be applied.
 */
public synchronized Settings validateUpdate(Settings settings) {
    final Settings current = Settings.builder().put(this.settings).put(settings).build();
    final Settings previous = Settings.builder().put(this.settings).put(this.lastSettingsApplied).build();
    List<RuntimeException> exceptions = new ArrayList<>();
    for (SettingUpdater<?> settingUpdater : settingUpdaters) {
        try {
            // ensure running this through the updater / dynamic validator
            // don't check if the value has changed we wanna test this anyways
            settingUpdater.getValue(current, previous);
        } catch (RuntimeException ex) {
            exceptions.add(ex);
            logger.debug(() -> new ParameterizedMessage("failed to prepareCommit settings for [{}]", settingUpdater), ex);
        }
    }
    // here we are exhaustive and record all settings that failed.
    ExceptionsHelper.rethrowAndSuppress(exceptions);
    return current;
}
 

public static String execute(Mustache template, Map<String, Object> params) {
    final StringWriter writer = new StringWriter();
    try {
        // crazy reflection here
        SecurityManager sm = System.getSecurityManager();
        if (sm != null) {
            sm.checkPermission(SPECIAL_PERMS);
        }
        AccessController.doPrivileged((PrivilegedAction<Void>) () -> {
            template.execute(writer, params);
            return null;
        });
    } catch (Exception e) {
        logger.error((Supplier<?>) () -> new ParameterizedMessage("Error running {}", template), e);
        throw new IllegalArgumentException("Error running " + template, e);
    }
    return writer.toString();

}
 

protected OpenNlpService start() {
    StopWatch sw = new StopWatch("models-loading");
    Map<String, String> settingsMap = IngestOpenNlpPlugin.MODEL_FILE_SETTINGS.getAsMap(settings);
    for (Map.Entry<String, String> entry : settingsMap.entrySet()) {
        String name = entry.getKey();
        sw.start(name);
        Path path = configDirectory.resolve(entry.getValue());
        try (InputStream is = Files.newInputStream(path)) {
            nameFinderModels.put(name, new TokenNameFinderModel(is));
        } catch (IOException e) {
            logger.error((Supplier<?>) () -> new ParameterizedMessage("Could not load model [{}] with path [{}]", name, path), e);
        }
        sw.stop();
    }

    if (settingsMap.keySet().size() == 0) {
        logger.error("Did not load any models for ingest-opennlp plugin, none configured");
    } else {
        logger.info("Read models in [{}] for {}", sw.totalTime(), settingsMap.keySet());
    }

    return this;
}
 

@Override
public PlayerData load(String username) throws Exception {
    PlayerData data = null;
    try (var connection = connectionPool.take();
         var loadData = connection.prepareStatement("SELECT json_data FROM main_data WHERE username = ?;")) {
        loadData.setString(1, username);

        try (var results = loadData.executeQuery()) {
            if (results.next()) {
                String jsonData = results.getString("json_data");
                data = Attribute.getGsonInstance().fromJson(jsonData, PlayerData.class);
            }
        }
    } catch (Exception e) {
        logger.warn(new ParameterizedMessage("{}'s data could not be loaded.", username), e);
    }
    return data;
}
 

@Override
public void onNodeAck(DiscoveryNode node, @Nullable Exception e) {
    if (node.equals(masterNode) == false && ackedTaskListener.mustAck(node) == false) {
        return;
    }
    if (e == null) {
        LOGGER.trace("ack received from node [{}], cluster_state update (version: {})", node, clusterStateVersion);
    } else {
        this.lastFailure = e;
        LOGGER.debug(() -> new ParameterizedMessage(
                "ack received from node [{}], cluster_state update (version: {})", node, clusterStateVersion), e);
    }

    if (countDown.countDown()) {
        finish();
    }
}
 

@VisibleForTesting
void updateJobSink(int size, TimeValue expiration) {
    LogSink<JobContextLog> sink = createSink(
        size, expiration, JOB_CONTEXT_LOG_ESTIMATOR, HierarchyCircuitBreakerService.JOBS_LOG);
    LogSink<JobContextLog> newSink = sink.equals(NoopLogSink.instance()) ? sink : new FilteredLogSink<>(
        memoryFilter,
        persistFilter,
        jobContextLog -> new ParameterizedMessage(
            "Statement execution: stmt=\"{}\" duration={}, user={} error=\"{}\"",
            jobContextLog.statement(),
            jobContextLog.ended() - jobContextLog.started(),
            jobContextLog.username(),
            jobContextLog.errorMessage()
        ),
        sink
    );
    jobsLogs.updateJobsLog(newSink);
}
 

@Test
public void testTraceEntryExit() {
    currentLevel = Level.TRACE;
    final FlowMessageFactory fact = new DefaultFlowMessageFactory();

    final ParameterizedMessage paramMsg = new ParameterizedMessage("Tracy {}", "Logan");
    currentEvent = new LogEvent(ENTRY_MARKER.getName(), fact.newEntryMessage(paramMsg), null);
    final EntryMessage entry = traceEntry("Tracy {}", "Logan");

    final ReusableParameterizedMessage msg = ReusableParameterizedMessageTest.set(
            new ReusableParameterizedMessage(), "Tracy {}", "Logan");
    ReusableParameterizedMessageTest.set(msg, "Some other message {}", 123);
    currentEvent = new LogEvent(null, msg, null);
    trace("Some other message {}", 123);

    // ensure original entry message not overwritten
    assertEquals("Tracy Logan", entry.getMessage().getFormattedMessage());

    currentEvent = new LogEvent(EXIT_MARKER.getName(), fact.newExitMessage(entry), null);
    traceExit(entry);

    // ensure original entry message not overwritten
    assertEquals("Tracy Logan", entry.getMessage().getFormattedMessage());
}
 

private void failAndRemoveShard(ShardRouting shardRouting, boolean sendShardFailure, String message, @Nullable Exception failure,
                                ClusterState state) {
    try {
        AllocatedIndex<? extends Shard> indexService = indicesService.indexService(shardRouting.shardId().getIndex());
        if (indexService != null) {
            indexService.removeShard(shardRouting.shardId().id(), message);
        }
    } catch (ShardNotFoundException e) {
        // the node got closed on us, ignore it
    } catch (Exception inner) {
        inner.addSuppressed(failure);
        LOGGER.warn(() -> new ParameterizedMessage(
                "[{}][{}] failed to remove shard after failure ([{}])",
                shardRouting.getIndexName(),
                shardRouting.getId(),
                message),
            inner);
    }
    if (sendShardFailure) {
        sendFailShard(shardRouting, message, failure, state);
    }
}
 

static void asyncLoggerTest(final Logger logger,
                            final Unlocker unlocker,
                            final BlockingAppender blockingAppender) {
    for (int i = 0; i < 130; i++) {
        TRACE("Test logging message " + i  + ". Remaining capacity=" + asyncRemainingCapacity(logger));
        TRACE("Test decrementing unlocker countdown latch. Count=" + unlocker.countDownLatch.getCount());
        unlocker.countDownLatch.countDown();
        final String param = "I'm innocent";
        logger.info(new ParameterizedMessage("logging innocent object #{} {}", i, param));
    }
    TRACE("Before stop() blockingAppender.logEvents.count=" + blockingAppender.logEvents.size());
    //CoreLoggerContexts.stopLoggerContext(false); // stop async thread
    while (blockingAppender.logEvents.size() < 130) { Thread.yield(); }
    TRACE("After  stop() blockingAppender.logEvents.count=" + blockingAppender.logEvents.size());

    final Stack<String> actual = transform(blockingAppender.logEvents);
    for (int i = 0; i < 130; i++) {
        assertEquals("logging innocent object #" + i + " I'm innocent", actual.pop());
    }
    assertTrue(actual.isEmpty());
}
 

@Test
public void testDisabledLookup() {
    final Configuration config = new DefaultConfigurationBuilder()
            .addProperty("foo", "bar")
            .build(true);
    final MessagePatternConverter converter = MessagePatternConverter.newInstance(
            config, new String[] {"nolookups"});
    final Message msg = new ParameterizedMessage("${foo}");
    final LogEvent event = Log4jLogEvent.newBuilder() //
            .setLoggerName("MyLogger") //
            .setLevel(Level.DEBUG) //
            .setMessage(msg).build();
    final StringBuilder sb = new StringBuilder();
    converter.format(event, sb);
    assertEquals("Expected the raw pattern string without lookup", "${foo}", sb.toString());
}
 

/**
 * Creates repository holder
 */
private Repository createRepository(RepositoryMetaData repositoryMetaData) {
    LOGGER.debug("creating repository [{}][{}]", repositoryMetaData.type(), repositoryMetaData.name());
    Repository.Factory factory = typesRegistry.get(repositoryMetaData.type());
    if (factory == null) {
        throw new RepositoryException(repositoryMetaData.name(),
            "repository type [" + repositoryMetaData.type() + "] does not exist");
    }
    try {
        Repository repository = factory.create(repositoryMetaData, typesRegistry::get);
        repository.start();
        return repository;
    } catch (Exception e) {
        LOGGER.warn(() -> new ParameterizedMessage("failed to create repository [{}][{}]", repositoryMetaData.type(), repositoryMetaData.name()), e);
        throw new RepositoryException(repositoryMetaData.name(), "failed to create repository", e);
    }
}
 

private void asyncCleanupUnlinkedShardLevelBlobs(SnapshotId snapshotId, Collection<ShardSnapshotMetaDeleteResult> deleteResults,
                                                 ActionListener<Void> listener) {
    threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(ActionRunnable.wrap(
        listener,
        l -> {
            try {
                blobContainer().deleteBlobsIgnoringIfNotExists(resolveFilesToDelete(snapshotId, deleteResults));
                l.onResponse(null);
            } catch (Exception e) {
                LOGGER.warn(
                    () -> new ParameterizedMessage("[{}] Failed to delete some blobs during snapshot delete", snapshotId),
                    e);
                throw e;
            }
        }));
}
 

static void asyncAppenderTest(final Logger logger,
                              final Unlocker unlocker,
                              final BlockingAppender blockingAppender) {
    for (int i = 0; i < 130; i++) {
        TRACE("Test logging message " + i  + ". Remaining capacity=" + asyncRemainingCapacity(logger));
        TRACE("Test decrementing unlocker countdown latch. Count=" + unlocker.countDownLatch.getCount());
        unlocker.countDownLatch.countDown();
        final String param = "I'm innocent";
        logger.info(new ParameterizedMessage("logging innocent object #{} {}", i, param));
    }
    TRACE("Before stop() blockingAppender.logEvents.count=" + blockingAppender.logEvents.size());
    //CoreLoggerContexts.stopLoggerContext(false); // stop async thread
    while (blockingAppender.logEvents.size() < 130) { Thread.yield(); }
    TRACE("After  stop() blockingAppender.logEvents.count=" + blockingAppender.logEvents.size());

    final Stack<String> actual = transform(blockingAppender.logEvents);
    for (int i = 0; i < 130; i++) {
        assertEquals("logging innocent object #" + i + " I'm innocent", actual.pop());
    }
    assertTrue(actual.isEmpty());
}
 

private void notifyListener(final GlobalCheckpointListener listener, final long globalCheckpoint, final Exception e) {
    assertNotification(globalCheckpoint, e);

    try {
        listener.accept(globalCheckpoint, e);
    } catch (final Exception caught) {
        if (globalCheckpoint != UNASSIGNED_SEQ_NO) {
            logger.warn(
                    new ParameterizedMessage(
                            "error notifying global checkpoint listener of updated global checkpoint [{}]",
                            globalCheckpoint),
                    caught);
        } else if (e instanceof IndexShardClosedException) {
            logger.warn("error notifying global checkpoint listener of closed shard", caught);
        } else {
            logger.warn("error notifying global checkpoint listener of timeout", caught);
        }
    }
}
 

public Set<String> children(String account, String container, BlobPath path) throws URISyntaxException, StorageException {
    final var blobsBuilder = new HashSet<String>();
    final Tuple<CloudBlobClient, Supplier<OperationContext>> client = client();
    final CloudBlobContainer blobContainer = client.v1().getContainerReference(container);
    final String keyPath = path.buildAsString();
    final EnumSet<BlobListingDetails> enumBlobListingDetails = EnumSet.of(BlobListingDetails.METADATA);

    for (ListBlobItem blobItem : blobContainer.listBlobs(keyPath, false, enumBlobListingDetails, null, client.v2().get())) {
        if (blobItem instanceof CloudBlobDirectory) {
            final URI uri = blobItem.getUri();
            LOGGER.trace(() -> new ParameterizedMessage("blob url [{}]", uri));
            // uri.getPath is of the form /container/keyPath.* and we want to strip off the /container/
            // this requires 1 + container.length() + 1, with each 1 corresponding to one of the /.
            // Lastly, we add the length of keyPath to the offset to strip this container's path.
            final String uriPath = uri.getPath();
            blobsBuilder.add(uriPath.substring(1 + container.length() + 1 + keyPath.length(), uriPath.length() - 1));
        }
    }
    return Set.copyOf(blobsBuilder);
}
 

static void asyncLoggerConfigTest(final Logger logger,
                                  final Unlocker unlocker,
                                  final BlockingAppender blockingAppender) {
    for (int i = 0; i < 130; i++) {
        TRACE("Test logging message " + i  + ". Remaining capacity=" + asyncRemainingCapacity(logger));
        TRACE("Test decrementing unlocker countdown latch. Count=" + unlocker.countDownLatch.getCount());
        unlocker.countDownLatch.countDown();
        final String param = "I'm innocent";
        logger.info(new ParameterizedMessage("logging innocent object #{} {}", i, param));
    }
    TRACE("Before stop() blockingAppender.logEvents.count=" + blockingAppender.logEvents.size());
    //CoreLoggerContexts.stopLoggerContext(false); // stop async thread
    while (blockingAppender.logEvents.size() < 130) { Thread.yield(); }
    TRACE("After  stop() blockingAppender.logEvents.count=" + blockingAppender.logEvents.size());

    final Stack<String> actual = transform(blockingAppender.logEvents);
    for (int i = 0; i < 130; i++) {
        assertEquals("logging innocent object #" + i + " I'm innocent", actual.pop());
    }
    assertTrue(actual.isEmpty());
}
 

@Override
public boolean sendPing() {
    for (TcpChannel channel : channels) {
        internalSendMessage(channel, pingMessage, new SendMetricListener(pingMessage.length()) {
            @Override
            protected void innerInnerOnResponse(Void v) {
                successfulPings.inc();
            }

            @Override
            protected void innerOnFailure(Exception e) {
                if (channel.isOpen()) {
                    logger.debug(() -> new ParameterizedMessage("[{}] failed to send ping transport message", node), e);
                    failedPings.inc();
                } else {
                    logger.trace(() ->
                        new ParameterizedMessage("[{}] failed to send ping transport message (channel closed)", node), e);
                }

            }
        });
    }
    return true;
}
 

@Test
public void testTraceEntryMessage() {
    currentLevel = Level.TRACE;
    final FlowMessageFactory fact = new DefaultFlowMessageFactory();

    final ParameterizedMessage paramMsg = new ParameterizedMessage("Tracy {}", "Logan");
    currentEvent = new LogEvent(ENTRY_MARKER.getName(), fact.newEntryMessage(paramMsg), null);

    final ReusableParameterizedMessage msg = ReusableParameterizedMessageTest.set(
            new ReusableParameterizedMessage(), "Tracy {}", "Logan");
    final EntryMessage entry = traceEntry(msg);

    ReusableParameterizedMessageTest.set(msg, "Some other message {}", 123);
    currentEvent = new LogEvent(null, msg, null);
    trace("Some other message {}", 123);

    // ensure original entry message not overwritten
    assertEquals("Tracy Logan", entry.getMessage().getFormattedMessage());

    currentEvent = new LogEvent(EXIT_MARKER.getName(), fact.newExitMessage(entry), null);
    traceExit(entry);

    // ensure original entry message not overwritten
    assertEquals("Tracy Logan", entry.getMessage().getFormattedMessage());
}
 

@Override
@SuppressForbidden(reason = "debug")
protected void stopInternal() {
    Releasables.close(() -> {
        final List<Tuple<String, Future<?>>> serverBootstrapCloseFutures = new ArrayList<>(serverBootstraps.size());
        for (final Map.Entry<String, ServerBootstrap> entry : serverBootstraps.entrySet()) {
            serverBootstrapCloseFutures.add(
                Tuple.tuple(entry.getKey(), entry.getValue().config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS)));
        }
        for (final Tuple<String, Future<?>> future : serverBootstrapCloseFutures) {
            future.v2().awaitUninterruptibly();
            if (!future.v2().isSuccess()) {
                logger.debug(
                    (Supplier<?>) () -> new ParameterizedMessage(
                        "Error closing server bootstrap for profile [{}]", future.v1()), future.v2().cause());
            }
        }
        serverBootstraps.clear();

        if (clientBootstrap != null) {
            clientBootstrap.config().group().shutdownGracefully(0, 5, TimeUnit.SECONDS).awaitUninterruptibly();
            clientBootstrap = null;
        }
    });
}
 

@Override
public Task unregister(Task task) {
    Task removedTask = super.unregister(task);
    if (removedTask != null) {
        for (MockTaskManagerListener listener : listeners) {
            try {
                listener.onTaskUnregistered(task);
            } catch (Exception e) {
                logger.warn(
                    (Supplier<?>) () -> new ParameterizedMessage(
                        "failed to notify task manager listener about unregistering the task with id {}", task.getId()), e);
            }
        }
    } else {
        logger.warn("trying to remove the same with id {} twice", task.getId());
    }
    return removedTask;
}
 

public Message convertEvent(final String message, final Object[] objects, final Throwable throwable) {
    try {
        final EventData data = objects != null && objects[0] instanceof EventData ?
                (EventData) objects[0] : new EventData(message);
        final StructuredDataMessage msg =
                new StructuredDataMessage(data.getEventId(), data.getMessage(), data.getEventType());
        for (final Map.Entry<String, Object> entry : data.getEventMap().entrySet()) {
            final String key = entry.getKey();
            if (EventData.EVENT_TYPE.equals(key) || EventData.EVENT_ID.equals(key)
                    || EventData.EVENT_MESSAGE.equals(key)) {
                continue;
            }
            msg.put(key, String.valueOf(entry.getValue()));
        }
        return msg;
    } catch (final Exception ex) {
        return new ParameterizedMessage(message, objects, throwable);
    }
}
 

private void updateMaxTermSeen(final long term) {
    synchronized (mutex) {
        maxTermSeen = Math.max(maxTermSeen, term);
        final long currentTerm = getCurrentTerm();
        if (mode == Mode.LEADER && maxTermSeen > currentTerm) {
            // Bump our term. However if there is a publication in flight then doing so would cancel the publication, so don't do that
            // since we check whether a term bump is needed at the end of the publication too.
            if (publicationInProgress()) {
                LOGGER.debug("updateMaxTermSeen: maxTermSeen = {} > currentTerm = {}, enqueueing term bump", maxTermSeen, currentTerm);
            } else {
                try {
                    LOGGER.debug("updateMaxTermSeen: maxTermSeen = {} > currentTerm = {}, bumping term", maxTermSeen, currentTerm);
                    ensureTermAtLeast(getLocalNode(), maxTermSeen);
                    startElection();
                } catch (Exception e) {
                    LOGGER.warn(new ParameterizedMessage("failed to bump term to {}", maxTermSeen), e);
                    becomeCandidate("updateMaxTermSeen");
                }
            }
        }
    }
}
 

@Override
protected void masterOperation(final UpdateSettingsRequest request, final ClusterState state, final ActionListener<AcknowledgedResponse> listener) {
    final Index[] concreteIndices = indexNameExpressionResolver.concreteIndices(state, request);
    UpdateSettingsClusterStateUpdateRequest clusterStateUpdateRequest = new UpdateSettingsClusterStateUpdateRequest()
            .indices(concreteIndices)
            .settings(request.settings())
            .setPreserveExisting(request.isPreserveExisting())
            .ackTimeout(request.timeout())
            .masterNodeTimeout(request.masterNodeTimeout());

    updateSettingsService.updateSettings(clusterStateUpdateRequest, new ActionListener<ClusterStateUpdateResponse>() {
        @Override
        public void onResponse(ClusterStateUpdateResponse response) {
            listener.onResponse(new AcknowledgedResponse(response.isAcknowledged()));
        }

        @Override
        public void onFailure(Exception t) {
            logger.debug(() -> new ParameterizedMessage("failed to update settings on indices [{}]", (Object) concreteIndices), t);
            listener.onFailure(t);
        }
    });
}
 

private void handlePredictionFailure(Exception e, String adID, String nodeID, AtomicReference<AnomalyDetectionException> failure) {
    LOG.error(new ParameterizedMessage("Received an error from node {} while fetching anomaly grade for {}", nodeID, adID), e);
    if (e == null) {
        return;
    }
    Throwable cause = ExceptionsHelper.unwrapCause(e);
    if (hasConnectionIssue(cause)) {
        handleConnectionException(nodeID);
    } else {
        findException(cause, adID, failure);
    }
}
 

private static String readResourceFile(String indexName, String resource) {
    try (InputStream is = IndexFeatureStore.class.getResourceAsStream(resource)) {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Streams.copy(is, out);
        return out.toString(StandardCharsets.UTF_8.name());
    } catch (Exception e) {
        LOGGER.error(
                (org.apache.logging.log4j.util.Supplier<?>) () -> new ParameterizedMessage(
                        "failed to create ltr feature store index [{}] with resource [{}]",
                        indexName, resource), e);
        throw new IllegalStateException("failed to create ltr feature store index with resource [" + resource + "]", e);
    }
}
 

@Override
public boolean isEnabled(final Level level, final Marker marker, final String message, final Object p0,
                         final Object p1, final Object p2, final Object p3,
                         final Object p4, final Object p5, final Object p6,
                         final Object p7, final Object p8) {
    return isEnabled(level, marker, new ParameterizedMessage(message, p0, p1, p2, p3, p4, p5, p6, p7, p8), null);
}
 

public Map<String, BlobMetaData> listBlobsByPrefix(String container, String keyPath, String prefix)
    throws URISyntaxException, StorageException {
    // NOTE: this should be here: if (prefix == null) prefix = "";
    // however, this is really inefficient since deleteBlobsByPrefix enumerates everything and
    // then does a prefix match on the result; it should just call listBlobsByPrefix with the prefix!
    final var blobsBuilder = new HashMap<String, BlobMetaData>();
    final EnumSet<BlobListingDetails> enumBlobListingDetails = EnumSet.of(BlobListingDetails.METADATA);
    final Tuple<CloudBlobClient, Supplier<OperationContext>> client = client();
    final CloudBlobContainer blobContainer = client.v1().getContainerReference(container);
    LOGGER.trace(() -> new ParameterizedMessage("listing container [{}], keyPath [{}], prefix [{}]", container, keyPath, prefix));
    for (final ListBlobItem blobItem : blobContainer.listBlobs(keyPath + (prefix == null ? "" : prefix), false,
        enumBlobListingDetails, null, client.v2().get())) {
        final URI uri = blobItem.getUri();
        LOGGER.trace(() -> new ParameterizedMessage("blob url [{}]", uri));
        // uri.getPath is of the form /container/keyPath.* and we want to strip off the /container/
        // this requires 1 + container.length() + 1, with each 1 corresponding to one of the /
        final String blobPath = uri.getPath().substring(1 + container.length() + 1);
        if (blobItem instanceof CloudBlob) {
            final BlobProperties properties = ((CloudBlob) blobItem).getProperties();
            final String name = blobPath.substring(keyPath.length());
            LOGGER.trace(() -> new ParameterizedMessage("blob url [{}], name [{}], size [{}]", uri, name, properties.getLength()));
            blobsBuilder.put(name, new PlainBlobMetaData(name, properties.getLength()));
        }
    }

    return Map.copyOf(blobsBuilder);
}