/**
 * Copyright 2016 Palantir Technologies
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.palantir.atlasdb.keyvalue.cassandra;

import java.util.Set;
import java.util.UUID;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.thrift.TException;

import com.google.common.annotations.VisibleForTesting;
import com.palantir.atlasdb.cassandra.CassandraKeyValueServiceConfig;
import com.palantir.atlasdb.keyvalue.api.TableReference;

public class SchemaMutationLockTables {
    public static final String LOCK_TABLE_PREFIX = "_locks";

    private static final Predicate<String> IS_LOCK_TABLE = table -> table.startsWith(LOCK_TABLE_PREFIX);

    private final CassandraClientPool clientPool;
    private final CassandraKeyValueServiceConfig config;

    public SchemaMutationLockTables(CassandraClientPool clientPool, CassandraKeyValueServiceConfig config) {
        this.clientPool = clientPool;
        this.config = config;
    }

    public Set<TableReference> getAllLockTables() throws TException {
        return clientPool.run(this::getAllLockTablesInternal);
    }

    private Set<TableReference> getAllLockTablesInternal(Cassandra.Client client) throws TException {
        return client.describe_keyspace(config.keyspace()).getCf_defs().stream()
                .map(CfDef::getName)
                .filter(IS_LOCK_TABLE)
                .map(TableReference::createUnsafe)
                .collect(Collectors.toSet());
    }

    public TableReference createLockTable() throws TException {
        return clientPool.run(this::createLockTable);
    }

    @VisibleForTesting
    protected TableReference createLockTable(UUID uuid) throws TException {
        return clientPool.run(client -> createInternalLockTable(client, uuid));
    }

    private TableReference createLockTable(Cassandra.Client client) throws TException {
        String lockTableName = LOCK_TABLE_PREFIX + "_" + getUniqueSuffix();
        TableReference lockTable = TableReference.createWithEmptyNamespace(lockTableName);
        createTableInternal(client, lockTable);
        return lockTable;
    }

    private String getUniqueSuffix() {
        return UUID.randomUUID().toString().replace('-','_');
    }

    private TableReference createInternalLockTable(Cassandra.Client client, UUID uuid) throws TException {
        String lockTableName = LOCK_TABLE_PREFIX + "_" + uuid.toString().replace('-','_');
        TableReference lockTable = TableReference.createWithEmptyNamespace(lockTableName);
        createTableInternal(client, lockTable);
        return lockTable;
    }

    private void createTableInternal(Cassandra.Client client, TableReference tableRef) throws TException {
        CfDef cf = CassandraConstants.getStandardCfDef(config.keyspace(), CassandraKeyValueService.internalTableName(tableRef));
        client.system_add_column_family(cf);
        CassandraKeyValueServices.waitForSchemaVersions(client, tableRef.getQualifiedName(), config.schemaMutationTimeoutMillis());
    }
}

@@ -882,64 +882,8 @@
 n;%0A%0A
-import com.google.common.annotations.VisibleForTesting;%0A
 impo
@@ -2138,194 +2138,8 @@
  %7D%0A%0A
-    @VisibleForTesting%0A    protected TableReference createLockTable(UUID uuid) throws TException %7B%0A        return clientPool.run(client -%3E createInternalLockTable(client, uuid));%0A    %7D%0A%0A
     
@@ -2214,32 +2214,32 @@
 ws TException %7B%0A
+
         String l
@@ -2521,430 +2521,146 @@
     
-return UUID.randomUUID().toString().
+// We 
 replace
-('-','_');%0A    %7D%0A%0A    private TableReference createInternalLockTable(Cassandra.Client client, UUID uuid) throws TException %7B%0A        String lockTableName = LOCK_TABLE_PREFIX + %22_%22 + uuid.toString().replace('-','_');%0A        TableReference lockTable = TableReference.createWithEmptyNamespace(lockTableName);%0A        createTableInternal(client, lockTable);%0A        return lockTable
+ '-' with '_' as hyphens are forbidden in Cassandra table names.%0A        return UUID.randomUUID().toString().replace('-','_')
 ;%0A  

Delete unused code; add helpful comment