This is a dummy implementation and this feature is not implemented
for Cassandra backed Conductor
548500
@Override
    public List<EventExecution> getEventExecutions(String eventHandlerName, String eventName, String messageId, int max) {
        throw new UnsupportedOperationException("This method is not implemented in CassandraExecutionDAO. Please use ExecutionDAOFacade instead.");
    }
49212
protected void createIndexesOnColumns(EntityMetadata m, String tableName, List<Column> columns, Class columnType) {
        Object pooledConnection = null;
        try {
            Cassandra.Client api = null;
            pooledConnection = getConnection();
            api = (org.apache.cassandra.thrift.Cassandra.Client) getConnection(pooledConnection);
            KsDef ksDef = api.describe_keyspace(m.getSchema());
            List<CfDef> cfDefs = ksDef.getCf_defs();

            // Column family definition on which secondary index creation is
            // required
            CfDef columnFamilyDefToUpdate = null;
            boolean isUpdatable = false;
            for (CfDef cfDef : cfDefs) {
                if (cfDef.getName().equals(tableName)) {
                    columnFamilyDefToUpdate = cfDef;
                    break;
                }
            }

            if (columnFamilyDefToUpdate == null) {
                log.error("Join table {} not available.", tableName);
                throw new PersistenceException("table" + tableName + " not found!");
            }
            // create a column family, in case it is not already available.

            // Get list of indexes already created
            List<ColumnDef> columnMetadataList = columnFamilyDefToUpdate.getColumn_metadata();
            List<String> indexList = new ArrayList<String>();

            if (columnMetadataList != null) {
                for (ColumnDef columnDef : columnMetadataList) {
                    indexList.add(new StringAccessor().fromBytes(String.class, columnDef.getName()));
                }
                // need to set them to null else it is giving problem on update
                // column family and trying to add again existing indexes.
                // columnFamilyDefToUpdate.column_metadata = null;
            }

            // Iterate over all columns for creating secondary index on them
            for (Column column : columns) {

                ColumnDef columnDef = new ColumnDef();

                columnDef.setName(column.getName());
                columnDef.setValidation_class(CassandraValidationClassMapper.getValidationClass(columnType, false));
                columnDef.setIndex_type(IndexType.KEYS);

                // Add secondary index only if it's not already created
                // (if already created, it would be there in column family
                // definition)
                if (!indexList.contains(new StringAccessor().fromBytes(String.class, column.getName()))) {
                    isUpdatable = true;
                    columnFamilyDefToUpdate.addToColumn_metadata(columnDef);
                }
            }

            // Finally, update column family with modified column family
            // definition
            if (isUpdatable) {
                columnFamilyDefToUpdate.setKey_validation_class(CassandraValidationClassMapper
                    .getValidationClass(m.getIdAttribute().getJavaType(), isCql3Enabled(m)));
                api.system_update_column_family(columnFamilyDefToUpdate);
            }

        } catch (Exception e) {
            log.warn("Could not create secondary index on column family {}, Caused by: . ", tableName, e);

        } finally {
            releaseConnection(pooledConnection);
        }
    }
265417
@SuppressWarnings("unchecked")
    @Override
    protected List<Object> recursivelyPopulateEntities(EntityMetadata m, Client client) {
        List<EnhanceEntity> ls = null;
        ApplicationMetadata appMetadata = kunderaMetadata.getApplicationMetadata();
        externalProperties = ((CassandraClientBase) client).getExternalProperties();
        MetamodelImpl metaModel =
            (MetamodelImpl) kunderaMetadata.getApplicationMetadata().getMetamodel(m.getPersistenceUnit());

        String query = appMetadata.getQuery(getJPAQuery());
        boolean isNative = kunderaQuery.isNative();

        if (isNative) {
            ls = (List<EnhanceEntity>) ((CassandraClientBase) client).executeQuery(m.getEntityClazz(),
                m.getRelationNames(), isNative, query != null ? query : getJPAQuery());
        } else if (!isNative && ((CassandraClientBase) client).isCql3Enabled(m)) {
            // edited
            // check if lucene or indexer are enabled then populate
            if (MetadataUtils.useSecondryIndex(((ClientBase) client).getClientMetadata())) {
                ls = ((CassandraClientBase) client).executeQuery(m.getEntityClazz(), m.getRelationNames(), isNative,

                    onQueryOverCQL3(m, client, metaModel, m.getRelationNames()));
            } else {
                ls = populateUsingLucene(m, client, null, getKunderaQuery().getResult());
            }
        } else {
            // Index in Inverted Index table if applicable
            boolean useInvertedIndex = CassandraIndexHelper.isInvertedIndexingApplicable(m,
                MetadataUtils.useSecondryIndex(((ClientBase) client).getClientMetadata()));
            Map<Boolean, List<IndexClause>> ixClause =
                MetadataUtils.useSecondryIndex(((ClientBase) client).getClientMetadata())
                    ? prepareIndexClause(m, useInvertedIndex) : null;

            if (useInvertedIndex && !getKunderaQuery().getFilterClauseQueue().isEmpty()) {
                ls = ((CassandraEntityReader) getReader()).readFromIndexTable(m, client, ixClause);
            } else {
                ((CassandraEntityReader) getReader()).setConditions(ixClause);
                ls = reader.populateRelation(m, client, isSingleResult ? 1 : this.maxResult);
            }
        }
        return setRelationEntities(ls, client, m);
    }
781477
public CassandraJavaPairRDD<K, V> toEmptyCassandraRDD() {
        CassandraRDD<Tuple2<K, V>> newRDD = rdd().toEmptyCassandraRDD();
        return wrap(newRDD);
    }
629726
@Override
  public String describeClusterName() throws HectorException {
    Operation<String> op = new Operation<String>(OperationType.META_READ, getCredentials()) {
      @Override
      public String execute(Cassandra.Client cassandra) throws HectorException {
        try {
          return cassandra.describe_cluster_name();
        } catch (Exception e) {
          throw xtrans.translate(e);
        }
      }
    };
    connectionManager.operateWithFailover(op);
    return op.getResult();
  }
954359
protected void persistJoinTableByCql(JoinTableData joinTableData, Cassandra.Client conn) {
        String joinTableName = joinTableData.getJoinTableName();
        String invJoinColumnName = joinTableData.getInverseJoinColumnName();
        Map<Object, Set<Object>> joinTableRecords = joinTableData.getJoinTableRecords();

        EntityMetadata entityMetadata =
            KunderaMetadataManager.getEntityMetadata(kunderaMetadata, joinTableData.getEntityClass());

        // need to bring in an insert query for this
        // add columns & execute query
        CQLTranslator translator = new CQLTranslator();

        String batch_Query = CQLTranslator.BATCH_QUERY;

        String insert_Query = translator.INSERT_QUERY;

        StringBuilder builder = new StringBuilder();
        builder.append(CQLTranslator.DEFAULT_KEY_NAME);
        builder.append(CQLTranslator.COMMA_STR);
        builder.append(translator.ensureCase(new StringBuilder(), joinTableData.getJoinColumnName(), false));
        builder.append(CQLTranslator.COMMA_STR);
        builder.append(translator.ensureCase(new StringBuilder(), joinTableData.getInverseJoinColumnName(), false));

        insert_Query = StringUtils.replace(insert_Query, CQLTranslator.COLUMN_FAMILY,
            translator.ensureCase(new StringBuilder(), joinTableName, false).toString());

        insert_Query = StringUtils.replace(insert_Query, CQLTranslator.COLUMNS, builder.toString());

        StringBuilder columnValueBuilder = new StringBuilder();

        StringBuilder statements = new StringBuilder();

        // insert query for each row key

        for (Object key : joinTableRecords.keySet()) {
            PropertyAccessor accessor =
                PropertyAccessorFactory.getPropertyAccessor((Field) entityMetadata.getIdAttribute().getJavaMember());

            Set<Object> values = joinTableRecords.get(key); // join column value

            for (Object value : values) {
                if (value != null) {
                    String insertQuery = insert_Query;
                    columnValueBuilder.append(CQLTranslator.QUOTE_STR);
                    columnValueBuilder.append(
                        PropertyAccessorHelper.getString(key) + "\001" + PropertyAccessorHelper.getString(value));
                    columnValueBuilder.append(CQLTranslator.QUOTE_STR);
                    columnValueBuilder.append(CQLTranslator.COMMA_STR);
                    translator.appendValue(columnValueBuilder, key.getClass(), key, true, false);
                    columnValueBuilder.append(CQLTranslator.COMMA_STR);
                    translator.appendValue(columnValueBuilder, value.getClass(), value, true, false);

                    insertQuery =
                        StringUtils.replace(insertQuery, CQLTranslator.COLUMN_VALUES, columnValueBuilder.toString());
                    statements.append(insertQuery);
                    statements.append(" ");
                }
            }
        }

        if (!StringUtils.isBlank(statements.toString())) {
            batch_Query = StringUtils.replace(batch_Query, CQLTranslator.STATEMENT, statements.toString());
            StringBuilder batchBuilder = new StringBuilder();
            batchBuilder.append(batch_Query);
            batchBuilder.append(CQLTranslator.APPLY_BATCH);
            execute(batchBuilder.toString(), conn);
        }

    }
548665
private <C> Iterator<Column<C>> columnScan(final ByteBuffer rowKey,
                                               final DeltaPlacement placement,
                                               final ColumnFamily<ByteBuffer, C> columnFamily,
                                               final C start,
                                               final C end,
                                               final boolean reversed,
                                               final ColumnInc<C> columnInc,
                                               final long limit,
                                               final long page,
                                               final ReadConsistency consistency) {
        return Iterators.concat(new AbstractIterator<Iterator<Column<C>>>() {
            private C _from = start;
            private long _remaining = limit;
            private long _page = page;

            @Override
            protected Iterator<Column<C>> computeNext() {
                if (_remaining <= 0) {
                    return endOfData();
                }

                // For page N+1, treat "_from" as exclusive.  Since Cassandra doesn't support exclusive column ranges
                // bump the from value up to the next possible time UUID (assumes from != null when page != 0).
                if (_page > 0) {
                    if (_from.equals(end)) {
                        return endOfData();
                    }
                    _from = reversed ? columnInc.previous(_from) : columnInc.next(_from);
                    if (_from == null) {
                        return endOfData();
                    }
                }

                // Execute the query
                int batchSize = (int) Math.min(_remaining, MAX_COLUMN_SCAN_BATCH);
                ColumnList<C> columns = execute(placement.getKeyspace()
                                .prepareQuery(columnFamily, SorConsistencies.toAstyanax(consistency))
                                .getKey(rowKey)
                                .withColumnRange(_from, end, reversed, batchSize),
                        "scan columns in placement %s, column family %s, row %s, from %s to %s",
                        placement.getName(), columnFamily.getName(), rowKey, start, end);

                // Update state for the next iteration.
                if (columns.size() >= batchSize) {
                    // Save the last column key so we can use it as the start (exclusive) if we must query to get more data.
                    _from = columns.getColumnByIndex(columns.size() - 1).getName();
                    _remaining = _remaining - columns.size();
                    _page++;
                } else {
                    // If we got fewer columns than we asked for, another query won't find more columns.
                    _remaining = 0;
                }

                // Track metrics.  For rows w/more than 50 columns, count subsequent reads w/_largeRowReadMeter.
                (_page == 0 ? _randomReadMeter : _largeRowReadMeter).mark();

                return columns.iterator();
            }
        });
    }
758972
private void addColumnToTable(TableInfo tableInfo, ColumnInfo column) throws Exception
    {
        CQLTranslator translator = new CQLTranslator();
        StringBuilder addColumnQuery = new StringBuilder("ALTER TABLE ");
        translator.ensureCase(addColumnQuery, tableInfo.getTableName(), false);
        addColumnQuery.append(" ADD ");
        translator.ensureCase(addColumnQuery, column.getColumnName(), false);
        addColumnQuery.append(" "
                + translator.getCQLType(CassandraValidationClassMapper.getValidationClass(column.getType(),
                        isCql3Enabled(tableInfo))));
        try
        {
            KunderaCoreUtils.printQuery(addColumnQuery.toString(), showQuery);
            cassandra_client.execute_cql3_query(ByteBuffer.wrap(addColumnQuery.toString().getBytes()),
                    Compression.NONE, ConsistencyLevel.ONE);
        }
        catch (InvalidRequestException ireforAddColumn)
        {
            StringBuilder ireforAddColumnbBuilder = new StringBuilder("Invalid column name ");
            ireforAddColumnbBuilder.append(column.getColumnName() + " because it conflicts with an existing column");
            if (ireforAddColumn.getWhy() != null && ireforAddColumn.getWhy().equals(ireforAddColumnbBuilder.toString()))
            {
                // alterColumnType(tableInfo, translator, column);
            }
            else
            {
                log.error("Error occurred while altering column type of  table {}, Caused by: .",
                        tableInfo.getTableName(), ireforAddColumn);
                throw new SchemaGenerationException("Error occurred while adding column into table "
                        + tableInfo.getTableName(), ireforAddColumn, "Cassandra", databaseName);
            }
        }

    }
886688
public void persistJoinTable(JoinTableData joinTableData)
    {
        Mutator mutator = clientFactory.getMutator(pool);

        String joinTableName = joinTableData.getJoinTableName();
        String invJoinColumnName = joinTableData.getInverseJoinColumnName();
        Map<Object, Set<Object>> joinTableRecords = joinTableData.getJoinTableRecords();
        EntityMetadata entityMetadata = KunderaMetadataManager.getEntityMetadata(kunderaMetadata,
                joinTableData.getEntityClass());

        if (isCql3Enabled(entityMetadata))
        {
            Cassandra.Client conn = null;
            Object pooledConnection = null;
            pooledConnection = getConnection();
            conn = (org.apache.cassandra.thrift.Cassandra.Client) getConnection(pooledConnection);
            persistJoinTableByCql(joinTableData, conn);
        }
        else
        {
            for (Object key : joinTableRecords.keySet())
            {
                Set<Object> values = joinTableRecords.get(key);

                List<Column> columns = new ArrayList<Column>();

                Class columnType = null;
                for (Object value : values)
                {
                    Column column = new Column();
                    column.setName(PropertyAccessorFactory.STRING.toBytes(invJoinColumnName
                            + Constants.JOIN_COLUMN_NAME_SEPARATOR + value.toString()));
                    column.setValue(PropertyAccessorHelper.getBytes(value));
                    column.setTimestamp(generator.getTimestamp());
                    columnType = value.getClass();
                    columns.add(column);
                }

                createIndexesOnColumns(entityMetadata, joinTableName, columns, columnType);

                mutator.writeColumns(joinTableName, Bytes.fromByteArray(PropertyAccessorHelper.getBytes(key)),
                        Arrays.asList(columns.toArray(new Column[0])));
            }
        }
        if (log.isInfoEnabled())
        {
            log.info(" Persisted data with join table column family {}", joinTableData.getJoinTableName());
        }
        mutator.execute(getConsistencyLevel());
    }
748715
@Override
    protected Object createPoolOrConnection()
    {
        PersistenceUnitMetadata persistenceUnitMetadata = kunderaMetadata.getApplicationMetadata()
                .getPersistenceUnitMetadata(getPersistenceUnit());

        Properties props = persistenceUnitMetadata.getProperties();

        String host = externalProperties != null ? (String) externalProperties.get(PersistenceProperties.KUNDERA_NODES)
                : null;
        String port = externalProperties != null ? (String) externalProperties.get(PersistenceProperties.KUNDERA_PORT)
                : null;

        if (host == null)
        {
            host = props.getProperty(PersistenceProperties.KUNDERA_NODES);
        }

        if (port == null)
        {
            port = props.getProperty(PersistenceProperties.KUNDERA_PORT);
        }

        String[] hosts = getHosts(host);

        Properties properties = ((ESClientPropertyReader) propertyReader).getConnectionProperties();

        Builder builder = Settings.settingsBuilder();

        builder.put("client.transport.sniff", true);

        if (properties != null)
        {
            builder.put(properties);
        }

        Settings settings = builder.build();

        TransportClient client = TransportClient.builder().settings(settings).build();

        for (String h : hosts)
        {
            client.addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress(h, new Integer(port))));
        }

        return client;
    }
720797
protected void setup()
    {
        try 
        {
            logger.info("Hostname: {}", InetAddress.getLocalHost().getHostName());
        }
        catch (UnknownHostException e1)
        {
            logger.info("Could not resolve local host");
        }

        long now = System.currentTimeMillis();
        if (now < EARLIEST_LAUNCH_DATE)
        {
            logger.error("current machine time is {}, but that is seemingly incorrect. exiting now.", new Date(now));
            System.exit(3);
        }

        // log warnings for different kinds of sub-optimal JVMs.  tldr use 64-bit Oracle >= 1.6u32
        if (!DatabaseDescriptor.hasLargeAddressSpace())
            logger.info("32bit JVM detected.  It is recommended to run Cassandra on a 64bit JVM for better performance.");
        String javaVersion = System.getProperty("java.version");
        String javaVmName = System.getProperty("java.vm.name");
        logger.info("JVM vendor/version: {}/{}", javaVmName, javaVersion);
        if (javaVmName.contains("OpenJDK"))
        {
            // There is essentially no QA done on OpenJDK builds, and
            // clusters running OpenJDK have seen many heap and load issues.
            logger.warn("OpenJDK is not recommended. Please upgrade to the newest Oracle Java release");
        }
        else if (!javaVmName.contains("HotSpot"))
        {
            logger.warn("Non-Oracle JVM detected.  Some features, such as immediate unmap of compacted SSTables, may not work as intended");
        }
     /*   else
        {
            String[] java_version = javaVersion.split("_");
            String java_major = java_version[0];
            int java_minor;
            try
            {
                java_minor = (java_version.length > 1) ? Integer.parseInt(java_version[1]) : 0;
            }
            catch (NumberFormatException e)
            {
                // have only seen this with java7 so far but no doubt there are other ways to break this
                logger.info("Unable to parse java version {}", Arrays.toString(java_version));
                java_minor = 32;
            }
        }
     */
        logger.info("Heap size: {}/{}", Runtime.getRuntime().totalMemory(), Runtime.getRuntime().maxMemory());
        for(MemoryPoolMXBean pool: ManagementFactory.getMemoryPoolMXBeans())
            logger.info("{} {}: {}", pool.getName(), pool.getType(), pool.getPeakUsage());
        logger.info("Classpath: {}", System.getProperty("java.class.path"));

        // Fail-fast if JNA is not available or failing to initialize properly
        // except with -Dcassandra.boot_without_jna=true. See CASSANDRA-6575.
        if (!CLibrary.jnaAvailable())
        {
            boolean jnaRequired = !Boolean.getBoolean("cassandra.boot_without_jna");

            if (jnaRequired)
            {
                logger.error("JNA failing to initialize properly. Use -Dcassandra.boot_without_jna=true to bootstrap even so.");
                System.exit(3);
            }
        }

        CLibrary.tryMlockall();

        maybeInitJmx();

        Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler()
        {
            public void uncaughtException(Thread t, Throwable e)
            {
                StorageMetrics.exceptions.inc();
                logger.error("Exception in thread {}", t, e);
                Tracing.trace("Exception in thread {}", t, e);
                for (Throwable e2 = e; e2 != null; e2 = e2.getCause())
                {
                    JVMStabilityInspector.inspectThrowable(e2);

                    if (e2 instanceof FSError)
                    {
                        if (e2 != e) // make sure FSError gets logged exactly once.
                            logger.error("Exception in thread {}", t, e2);
                        FileUtils.handleFSError((FSError) e2);
                    }

                    if (e2 instanceof CorruptSSTableException)
                    {
                        if (e2 != e)
                            logger.error("Exception in thread " + t, e2);
                        FileUtils.handleCorruptSSTable((CorruptSSTableException) e2);
                    }
                }
            }
        });

        // check all directories(data, commitlog, saved cache) for existence and permission
        Iterable<String> dirs = Iterables.concat(Arrays.asList(DatabaseDescriptor.getAllDataFileLocations()),
                                                 Arrays.asList(DatabaseDescriptor.getCommitLogLocation(),
                                                               DatabaseDescriptor.getSavedCachesLocation()));
        for (String dataDir : dirs)
        {
            logger.debug("Checking directory {}", dataDir);
            File dir = new File(dataDir);

            // check that directories exist.
            if (!dir.exists())
            {
                logger.error("Directory {} doesn't exist", dataDir);
                // if they don't, failing their creation, stop cassandra.
                if (!dir.mkdirs())
                {
                    logger.error("Has no permission to create {} directory", dataDir);
                    System.exit(3);
                }
            }
            // if directories exist verify their permissions
            if (!Directories.verifyFullPermissions(dir, dataDir))
            {
                // if permissions aren't sufficient, stop cassandra.
                System.exit(3);
            }
        }

        if (CacheService.instance == null) // should never happen
            throw new RuntimeException("Failed to initialize Cache Service.");

        // check the system keyspace to keep user from shooting self in foot by changing partitioner, cluster name, etc.
        // we do a one-off scrub of the system keyspace first; we can't load the list of the rest of the keyspaces,
        // until system keyspace is opened.
        for (CFMetaData cfm : Schema.instance.getKeyspaceMetaData(Keyspace.SYSTEM_KS).values())
            ColumnFamilyStore.scrubDataDirectories(cfm);
        try
        {
            SystemKeyspace.checkHealth();
        }
        catch (ConfigurationException e)
        {
            logger.error("Fatal exception during initialization", e);
            System.exit(100);
        }

        // load keyspace descriptions.
        DatabaseDescriptor.loadSchemas();

        // clean up compaction leftovers
        Map<Pair<String, String>, Map<Integer, UUID>> unfinishedCompactions = SystemKeyspace.getUnfinishedCompactions();
        for (Pair<String, String> kscf : unfinishedCompactions.keySet())
        {
            CFMetaData cfm = Schema.instance.getCFMetaData(kscf.left, kscf.right);
            // CFMetaData can be null if CF is already dropped
            if (cfm != null)
                ColumnFamilyStore.removeUnfinishedCompactionLeftovers(cfm, unfinishedCompactions.get(kscf));
        }
        SystemKeyspace.discardCompactionsInProgress();

        // clean up debris in the rest of the keyspaces
        for (String keyspaceName : Schema.instance.getKeyspaces())
        {
            // Skip system as we've already cleaned it
            if (keyspaceName.equals(Keyspace.SYSTEM_KS))
                continue;

            for (CFMetaData cfm : Schema.instance.getKeyspaceMetaData(keyspaceName).values())
                ColumnFamilyStore.scrubDataDirectories(cfm);
        }

        Keyspace.setInitialized();
        // initialize keyspaces
        for (String keyspaceName : Schema.instance.getKeyspaces())
        {
            if (logger.isDebugEnabled())
                logger.debug("opening keyspace {}", keyspaceName);
            // disable auto compaction until commit log replay ends
            for (ColumnFamilyStore cfs : Keyspace.open(keyspaceName).getColumnFamilyStores())
            {
                for (ColumnFamilyStore store : cfs.concatWithIndexes())
                {
                    store.disableAutoCompaction();
                }
            }
        }

        if (CacheService.instance.keyCache.size() > 0)
            logger.info("completed pre-loading ({} keys) key cache.", CacheService.instance.keyCache.size());

        if (CacheService.instance.rowCache.size() > 0)
            logger.info("completed pre-loading ({} keys) row cache.", CacheService.instance.rowCache.size());

        try
        {
            GCInspector.register();
        }
        catch (Throwable t)
        {
            JVMStabilityInspector.inspectThrowable(t);
            logger.warn("Unable to start GCInspector (currently only supported on the Sun JVM)");
        }

        // replay the log if necessary
        try
        {
            CommitLog.instance.recover();
        }
        catch (IOException e)
        {
            throw new RuntimeException(e);
        }

        // enable auto compaction
        for (Keyspace keyspace : Keyspace.all())
        {
            for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
            {
                for (final ColumnFamilyStore store : cfs.concatWithIndexes())
                {
                    if (store.getCompactionStrategy().shouldBeEnabled())
                        store.enableAutoCompaction();
                }
            }
        }
        // start compactions in five minutes (if no flushes have occurred by then to do so)
        Runnable runnable = new Runnable()
        {
            public void run()
            {
                for (Keyspace keyspaceName : Keyspace.all())
                {
                    for (ColumnFamilyStore cf : keyspaceName.getColumnFamilyStores())
                    {
                        for (ColumnFamilyStore store : cf.concatWithIndexes())
                            CompactionManager.instance.submitBackground(store);
                    }
                }
            }
        };
        ScheduledExecutors.optionalTasks.schedule(runnable, 5, TimeUnit.MINUTES);

        SystemKeyspace.finishStartup();

        // start server internals
        StorageService.instance.registerDaemon(this);
        try
        {
            StorageService.instance.initServer();
        }
        catch (ConfigurationException e)
        {
            logger.error("Fatal configuration error", e);
            System.err.println(e.getMessage() + "\nFatal configuration error; unable to start server.  See log for stacktrace.");
            System.exit(1);
        }

        Mx4jTool.maybeLoad();

        // Metrics
        String metricsReporterConfigFile = System.getProperty("cassandra.metricsReporterConfigFile");
        if (metricsReporterConfigFile != null)
        {
            logger.info("Trying to load metrics-reporter-config from file: {}", metricsReporterConfigFile);
            try
            {
                String reportFileLocation = CassandraDaemon.class.getClassLoader().getResource(metricsReporterConfigFile).getFile();
                ReporterConfig.loadFromFile(reportFileLocation).enableAll();
            }
            catch (Exception e)
            {
                logger.warn("Failed to load metrics-reporter-config, metric sinks will not be activated", e);
            }
        }

        if (!FBUtilities.getBroadcastAddress().equals(InetAddress.getLoopbackAddress()))
            waitForGossipToSettle();

        // schedule periodic dumps of table size estimates into SystemKeyspace.SIZE_ESTIMATES_CF
        // set cassandra.size_recorder_interval to 0 to disable
        int sizeRecorderInterval = Integer.getInteger("cassandra.size_recorder_interval", 5 * 60);
        if (sizeRecorderInterval > 0)
            ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(SizeEstimatesRecorder.instance, 30, sizeRecorderInterval, TimeUnit.SECONDS);

        // Thrift
        InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress();
        int rpcPort = DatabaseDescriptor.getRpcPort();
        int listenBacklog = DatabaseDescriptor.getRpcListenBacklog();
        thriftServer = new ThriftServer(rpcAddr, rpcPort, listenBacklog);

        // Native transport
        InetAddress nativeAddr = DatabaseDescriptor.getRpcAddress();
        int nativePort = DatabaseDescriptor.getNativeTransportPort();
        nativeServer = new org.apache.cassandra.transport.Server(nativeAddr, nativePort);
    }
644483
public boolean isCql3Enabled(EntityMetadata metadata) {
        if (metadata != null) {

            MetamodelImpl metaModel =
                (MetamodelImpl) kunderaMetadata.getApplicationMetadata().getMetamodel(metadata.getPersistenceUnit());

            if (metaModel.isEmbeddable(metadata.getIdAttribute().getBindableJavaType())) {
                return true;
            }
            // added for embeddables support on cql3
            AbstractManagedType managedType = (AbstractManagedType) metaModel.entity(metadata.getEntityClazz());
            if (managedType.hasEmbeddableAttribute()) {
                return getCqlVersion().equalsIgnoreCase(CassandraConstants.CQL_VERSION_3_0);
            }

            if (getCqlVersion().equalsIgnoreCase(CassandraConstants.CQL_VERSION_3_0)
                && metadata.getType().equals(Type.SUPER_COLUMN_FAMILY)) {
                log.warn(
                    "Super Columns not supported by cql, Any operation on supercolumn family will be executed using thrift, returning false.");
                return false;
            }
            return getCqlVersion().equalsIgnoreCase(CassandraConstants.CQL_VERSION_3_0);
        }
        return getCqlVersion().equalsIgnoreCase(CassandraConstants.CQL_VERSION_3_0);
    }
1207702
public boolean addCassandraHost(CassandraHost cassandraHost) {
    if ( !getHosts().contains(cassandraHost) ) {
      HClientPool pool = null;
      try {
        cassandraHostConfigurator.applyConfig(cassandraHost);
        pool = cassandraHostConfigurator.getLoadBalancingPolicy().createConnection(clientFactory, cassandraHost, monitor);
        hostPools.putIfAbsent(cassandraHost, pool);
        log.info("Added host {} to pool", cassandraHost.getName());
        listenerHandler.fireOnAddHost(cassandraHost, true, null, null);
        return true;
      } catch (HectorTransportException hte) {
        String errorMessage = "Transport exception host to HConnectionManager: " + cassandraHost;
        log.error(errorMessage, hte);
        listenerHandler.fireOnAddHost(cassandraHost, false, errorMessage, hte);
      } catch (Exception ex) {
        String errorMessage = "General exception host to HConnectionManager: " + cassandraHost;
        log.error(errorMessage, ex);
        listenerHandler.fireOnAddHost(cassandraHost, false, errorMessage, ex);
      }
    } else {
      String message = "Host already existed for pool " + cassandraHost.getName();
      log.info(message);
      listenerHandler.fireOnAddHost(cassandraHost, false, message, null);
    }
    return false;
  }
1251286
public static <IN> CassandraSinkBuilder<IN> addSink(DataStream<IN> input) {
		TypeInformation<IN> typeInfo = input.getType();
		if (typeInfo instanceof TupleTypeInfo) {
			DataStream<Tuple> tupleInput = (DataStream<Tuple>) input;
			return (CassandraSinkBuilder<IN>) new CassandraTupleSinkBuilder<>(tupleInput, tupleInput.getType(), tupleInput.getType().createSerializer(tupleInput.getExecutionEnvironment().getConfig()));
		}
		if (typeInfo instanceof RowTypeInfo) {
			DataStream<Row> rowInput = (DataStream<Row>) input;
			return (CassandraSinkBuilder<IN>) new CassandraRowSinkBuilder(rowInput, rowInput.getType(), rowInput.getType().createSerializer(rowInput.getExecutionEnvironment().getConfig()));
		}
		if (typeInfo instanceof PojoTypeInfo) {
			return new CassandraPojoSinkBuilder<>(input, input.getType(), input.getType().createSerializer(input.getExecutionEnvironment().getConfig()));
		}
		if (typeInfo instanceof CaseClassTypeInfo) {
			DataStream<Product> productInput = (DataStream<Product>) input;
			return (CassandraSinkBuilder<IN>) new CassandraScalaProductSinkBuilder<>(productInput, productInput.getType(), productInput.getType().createSerializer(input.getExecutionEnvironment().getConfig()));
		}
		throw new IllegalArgumentException("No support for the type of the given DataStream: " + input.getType());
	}
769938
public int executeBatch() {
        Cassandra.Client conn = null;
        Object pooledConnection = null;

        /**
         * Key -> Entity Class Value -> Map containing Row ID as Key and Mutation List as Value
         */
        Map<Class<?>, Map<ByteBuffer, Map<String, List<Mutation>>>> batchMutationMap =
            new HashMap<Class<?>, Map<ByteBuffer, Map<String, List<Mutation>>>>();

        int recordsExecuted = 0;
        boolean setCounter = true;
        String batchQuery = CQLTranslator.BATCH_QUERY;
        batchQuery = StringUtils.replace(batchQuery, CQLTranslator.STATEMENT, "");
        StringBuilder batchQueryBuilder = new StringBuilder(batchQuery);
        try {
            boolean isCql3Enabled = false;
            for (Node node : nodes) {
                if (node.isDirty()) {
                    node.handlePreEvent();
                    Object entity = node.getData();
                    Object id = node.getEntityId();
                    EntityMetadata metadata =
                        KunderaMetadataManager.getEntityMetadata(kunderaMetadata, node.getDataClass());
                    if (metadata.isCounterColumnType() && setCounter) {
                        batchQueryBuilder = new StringBuilder(StringUtils.replace(batchQueryBuilder.toString(),
                            CQLTranslator.BEGIN_BATCH, CQLTranslator.BEGIN_COUNTER_BATCH));
                        setCounter = false;
                    }
                    persistenceUnit = metadata.getPersistenceUnit();
                    isUpdate = node.isUpdate();

                    MetamodelImpl metaModel = (MetamodelImpl) kunderaMetadata.getApplicationMetadata()
                        .getMetamodel(metadata.getPersistenceUnit());

                    // delete can not be executed in batch

                    if (isCql3Enabled(metadata)) {
                        isCql3Enabled = true;
                        List<RelationHolder> relationHolders = getRelationHolders(node);
                        if (node.isInState(RemovedState.class)) {
                            String query;
                            query = onDeleteQuery(metadata, metadata.getTableName(), metaModel, id);
                            batchQueryBuilder.append(Constants.SPACE);
                            batchQueryBuilder.append(query);
                        } else {
                            List<String> insertQueries = getPersistQueries(metadata, entity, conn, relationHolders,
                                getTtlValues().get(metadata.getTableName()));
                            for (String query : insertQueries) {
                                batchQueryBuilder.append(Constants.SPACE);
                                batchQueryBuilder.append(query);
                            }
                        }
                    } else {
                        if (node.isInState(RemovedState.class)) {
                            delete(entity, id);
                        } else {
                            List<RelationHolder> relationHolders = getRelationHolders(node);
                            Map<ByteBuffer, Map<String, List<Mutation>>> mutationMap =
                                new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
                            mutationMap = prepareMutation(metadata, entity, id, relationHolders, mutationMap);

                            recordsExecuted += mutationMap.size();
                            if (!batchMutationMap.containsKey(metadata.getEntityClazz())) {
                                batchMutationMap.put(metadata.getEntityClazz(), mutationMap);
                            } else {
                                batchMutationMap.get(metadata.getEntityClazz()).putAll(mutationMap);
                            }

                            indexNode(node, metadata);
                        }
                    }
                    node.handlePostEvent();
                }
            }

            // Write Mutation map to database

            if (!batchMutationMap.isEmpty()) {
                pooledConnection = getConnection();
                conn = (org.apache.cassandra.thrift.Cassandra.Client) getConnection(pooledConnection);

                for (Class<?> entityClass : batchMutationMap.keySet()) {
                    conn.batch_mutate(batchMutationMap.get(entityClass), consistencyLevel);
                }
            }

            if (!nodes.isEmpty() && isCql3Enabled) {
                batchQueryBuilder.append(CQLTranslator.APPLY_BATCH);
                executeCQLQuery(batchQueryBuilder.toString(), isCql3Enabled);
            }
        } catch (Exception e) {
            log.error("Error while persisting record. Caused by: .", e);
            throw new KunderaException(e);
        } finally {
            clear();
            if (pooledConnection != null) {
                releaseConnection(pooledConnection);
            }
        }

        return recordsExecuted;
    }
917632
private void doDeliverHintsToEndpoint(InetAddress endpoint)
    {
        // find the hints for the node using its token.
        UUID hostId = Gossiper.instance.getHostId(endpoint);
        logger.info("Started hinted handoff for host: {} with IP: {}", hostId, endpoint);
        final ByteBuffer hostIdBytes = ByteBuffer.wrap(UUIDGen.decompose(hostId));
        DecoratedKey epkey =  StorageService.getPartitioner().decorateKey(hostIdBytes);

        final AtomicInteger rowsReplayed = new AtomicInteger(0);
        Composite startColumn = Composites.EMPTY;

        int pageSize = calculatePageSize();
        logger.debug("Using pageSize of {}", pageSize);

        // rate limit is in bytes per second. Uses Double.MAX_VALUE if disabled (set to 0 in cassandra.yaml).
        // max rate is scaled by the number of nodes in the cluster (CASSANDRA-5272).
        int throttleInKB = DatabaseDescriptor.getHintedHandoffThrottleInKB()
                           / (StorageService.instance.getTokenMetadata().getAllEndpoints().size() - 1);
        RateLimiter rateLimiter = RateLimiter.create(throttleInKB == 0 ? Double.MAX_VALUE : throttleInKB * 1024);

        delivery:
        while (true)
        {
            long now = System.currentTimeMillis();
            QueryFilter filter = QueryFilter.getSliceFilter(epkey,
                                                            SystemKeyspace.HINTS_CF,
                                                            startColumn,
                                                            Composites.EMPTY,
                                                            false,
                                                            pageSize,
                                                            now);

            ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), (int) (now / 1000));

            if (pagingFinished(hintsPage, startColumn))
            {
                logger.info("Finished hinted handoff of {} rows to endpoint {}", rowsReplayed, endpoint);
                break;
            }

            // check if node is still alive and we should continue delivery process
            if (!FailureDetector.instance.isAlive(endpoint))
            {
                logger.info("Endpoint {} died during hint delivery; aborting ({} delivered)", endpoint, rowsReplayed);
                break;
            }

            List<WriteResponseHandler> responseHandlers = Lists.newArrayList();
            for (final Cell hint : hintsPage)
            {
                // check if hints delivery has been paused during the process
                if (hintedHandOffPaused)
                {
                    logger.debug("Hints delivery process is paused, aborting");
                    break delivery;
                }

                // Skip tombstones:
                // if we iterate quickly enough, it's possible that we could request a new page in the same millisecond
                // in which the local deletion timestamp was generated on the last column in the old page, in which
                // case the hint will have no columns (since it's deleted) but will still be included in the resultset
                // since (even with gcgs=0) it's still a "relevant" tombstone.
                if (!hint.isLive())
                    continue;

                startColumn = hint.name();

                int version = Int32Type.instance.compose(hint.name().get(1));
                DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(hint.value()));
                Mutation mutation;
                try
                {
                    mutation = Mutation.serializer.deserialize(in, version);
                }
                catch (UnknownColumnFamilyException e)
                {
                    logger.debug("Skipping delivery of hint for deleted columnfamily", e);
                    deleteHint(hostIdBytes, hint.name(), hint.timestamp());
                    continue;
                }
                catch (IOException e)
                {
                    throw new AssertionError(e);
                }

                for (UUID cfId : mutation.getColumnFamilyIds())
                {
                    if (hint.timestamp() <= SystemKeyspace.getTruncatedAt(cfId))
                    {
                        logger.debug("Skipping delivery of hint for truncated columnfamily {}", cfId);
                        mutation = mutation.without(cfId);
                    }
                }

                if (mutation.isEmpty())
                {
                    deleteHint(hostIdBytes, hint.name(), hint.timestamp());
                    continue;
                }

                MessageOut<Mutation> message = mutation.createMessage();
                rateLimiter.acquire(message.serializedSize(MessagingService.current_version));
                Runnable callback = new Runnable()
                {
                    public void run()
                    {
                        rowsReplayed.incrementAndGet();
                        deleteHint(hostIdBytes, hint.name(), hint.timestamp());
                    }
                };
                WriteResponseHandler responseHandler = new WriteResponseHandler(endpoint, WriteType.SIMPLE, callback);
                MessagingService.instance().sendRR(message, endpoint, responseHandler, false);
                responseHandlers.add(responseHandler);
            }

            for (WriteResponseHandler handler : responseHandlers)
            {
                try
                {
                    handler.get();
                }
                catch (WriteTimeoutException e)
                {
                    logger.info("Timed out replaying hints to {}; aborting ({} delivered)", endpoint, rowsReplayed);
                    break delivery;
                }
            }
        }

        // Flush all the tombstones to disk
        hintStore.forceBlockingFlush();
    }
456610
private void onElementCollectionColumns(CQLTranslator translator, List<CollectionColumnInfo> collectionColumnInfos,
            StringBuilder queryBuilder)
    {
        for (CollectionColumnInfo cci : collectionColumnInfos)
        {
            String dataType = CassandraValidationClassMapper.getValidationClass(cci.getType(), true);

            // CQL Type of collection column
            String collectionCqlType = translator.getCQLType(dataType);

            // Collection Column Name
            String collectionColumnName = new String(cci.getCollectionColumnName());

            // Generic Type list
            StringBuilder genericTypesBuilder = null;
            List<Class<?>> genericClasses = cci.getGenericClasses();
            if (!genericClasses.isEmpty())
            {
                genericTypesBuilder = new StringBuilder();
                if (MapType.class.getSimpleName().equals(dataType) && genericClasses.size() == 2)
                {
                    genericTypesBuilder.append(Constants.STR_LT);
                    if (genericClasses.get(0).getAnnotation(Embeddable.class) != null)
                    {
                        String frozenKey = CQLTranslator.FROZEN + Constants.STR_LT + Constants.ESCAPE_QUOTE
                                + genericClasses.get(0).getSimpleName() + Constants.ESCAPE_QUOTE + Constants.STR_GT;
                        genericTypesBuilder.append(frozenKey);
                    }
                    else
                    {
                        String keyDataType = CassandraValidationClassMapper.getValidationClass(genericClasses.get(0),
                                true);
                        genericTypesBuilder.append(translator.getCQLType(keyDataType));
                    }
                    genericTypesBuilder.append(Constants.SPACE_COMMA);
                    if (genericClasses.get(1).getAnnotation(Embeddable.class) != null)
                    {
                        String frozenKey = CQLTranslator.FROZEN + Constants.STR_LT + Constants.ESCAPE_QUOTE
                                + genericClasses.get(1).getSimpleName() + Constants.ESCAPE_QUOTE + Constants.STR_GT;
                        genericTypesBuilder.append(frozenKey);
                    }
                    else
                    {
                        String keyDataType = CassandraValidationClassMapper.getValidationClass(genericClasses.get(1),
                                true);
                        genericTypesBuilder.append(translator.getCQLType(keyDataType));
                    }
                    genericTypesBuilder.append(Constants.STR_GT);
                }
                else if ((ListType.class.getSimpleName().equals(dataType) || SetType.class.getSimpleName().equals(
                        dataType))
                        && genericClasses.size() == 1)
                {
                    genericTypesBuilder.append(Constants.STR_LT);
                    if (genericClasses.get(0).getAnnotation(Embeddable.class) != null)
                    {
                        String frozenKey = CQLTranslator.FROZEN + Constants.STR_LT + Constants.ESCAPE_QUOTE
                                + genericClasses.get(0).getSimpleName() + Constants.ESCAPE_QUOTE + Constants.STR_GT;
                        genericTypesBuilder.append(frozenKey);
                    }
                    else
                    {
                        String keyDataType = CassandraValidationClassMapper.getValidationClass(genericClasses.get(0),
                                true);
                        genericTypesBuilder.append(translator.getCQLType(keyDataType));
                    }
                    genericTypesBuilder.append(Constants.STR_GT);
                }
                else
                {
                    throw new SchemaGenerationException("Incorrect collection field definition for "
                            + cci.getCollectionColumnName() + ". Generic Types must be defined correctly.");
                }
            }

            if (genericTypesBuilder != null)
            {
                collectionCqlType += genericTypesBuilder.toString();
            }

            translator.appendColumnName(queryBuilder, collectionColumnName, collectionCqlType);
            queryBuilder.append(Constants.SPACE_COMMA);

        }

    }