public void verify(
        Optional<String> expectedClusterId,
        OptionalInt expectedNodeId,
        EnumSet<VerificationFlag> verificationFlags
    ) {
        Map<Uuid, String> seenUuids = new HashMap<>();
        if (verificationFlags.contains(VerificationFlag.REQUIRE_AT_LEAST_ONE_VALID)) {
            if (logDirProps.isEmpty()) {
                throw new RuntimeException("No readable meta.properties files found.");
            }
        }
        for (Entry<String, MetaProperties> entry : logDirProps.entrySet()) {
            String logDir = entry.getKey();
            String path = new File(logDir, META_PROPERTIES_NAME).toString();
            MetaProperties metaProps = entry.getValue();
            if (verificationFlags.contains(VerificationFlag.REQUIRE_V0)) {
                if (!metaProps.version().equals(MetaPropertiesVersion.V0)) {
                    throw new RuntimeException("Found unexpected version in " + path + ". " +
                        "ZK-based brokers that are not migrating only support version 0 " +
                        "(which is implicit when the `version` field is missing).");
                }
            }
            if (!metaProps.clusterId().isPresent()) {
                if (metaProps.version().alwaysHasClusterId()) {
                    throw new RuntimeException("cluster.id was not specified in the v1 file: " +
                        path);
                }
            } else if (!expectedClusterId.isPresent()) {
                expectedClusterId = metaProps.clusterId();
            } else if (!metaProps.clusterId().get().equals(expectedClusterId.get())) {
                throw new RuntimeException("Invalid cluster.id in: " + path + ". Expected " +
                    expectedClusterId.get() + ", but read " + metaProps.clusterId().get());
            }
            if (!metaProps.nodeId().isPresent()) {
                if (metaProps.version().alwaysHasNodeId()) {
                    throw new RuntimeException("node.id was not specified in " + path);
                }
            } else if (!expectedNodeId.isPresent()) {
                expectedNodeId = metaProps.nodeId();
            } else if (metaProps.nodeId().getAsInt() != expectedNodeId.getAsInt()) {
                throw new RuntimeException("Stored node id " + metaProps.nodeId().getAsInt() +
                    " doesn't match previous node id " + expectedNodeId.getAsInt() + " in " + path +
                    ". If you moved your data, make sure your configured node id matches. If you " +
                    "intend to create a new node, you should remove all data in your data " +
                    "directories.");
            }
            if (metaProps.directoryId().isPresent()) {
                if (DirectoryId.reserved(metaProps.directoryId().get())) {
                    throw new RuntimeException("Invalid resrved directory ID " +
                        metaProps.directoryId().get() + " found in " + logDir);
                }
                String prevLogDir = seenUuids.put(metaProps.directoryId().get(), logDir);
                if (prevLogDir != null) {
                    throw new RuntimeException("Duplicate directory ID " + metaProps.directoryId() +
                        " found. It was the ID of " + prevLogDir + ", " + "but also of " +
                        logDir);
                }
            }
        }
        if (verificationFlags.contains(VerificationFlag.REQUIRE_METADATA_LOG_DIR)) {
            if (!metadataLogDir.isPresent()) {
                throw new RuntimeException("No metadata log directory was specified.");
            }
        }
        if (metadataLogDir.isPresent()) {
            if (errorLogDirs.contains(metadataLogDir.get())) {
                throw new RuntimeException("Encountered I/O error in metadata log directory " +
                        metadataLogDir.get() + ". Cannot continue.");
            }
        }
    }
@Test
    public void testVerificationFailureOnLackOfMetadataLogDir() {
        MetaPropertiesEnsemble ensemble = new MetaPropertiesEnsemble(
            Collections.singleton("/tmp/foo1"),
            Collections.emptySet(),
            Collections.emptyMap(),
            Optional.empty());
        assertEquals("No metadata log directory was specified.",
            assertThrows(RuntimeException.class,
                () -> ensemble.verify(Optional.empty(),
                    OptionalInt.empty(),
                    EnumSet.of(REQUIRE_METADATA_LOG_DIR))).
                        getMessage());
    }