@Override
    public void persistJoinTable(JoinTableData joinTableData)
    {
        String joinTableName = joinTableData.getJoinTableName();
        String invJoinColumnName = joinTableData.getInverseJoinColumnName();
        Map<Object, Set<Object>> joinTableRecords = joinTableData.getJoinTableRecords();

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

        Connection conn = null;
        try
        {

            conn = getConnection();

            if (isCql3Enabled(entityMetadata))
            {
                persistJoinTableByCql(joinTableData, conn.getClient());
            }
            else
            {
                KunderaCoreUtils.printQuery("Persist join table:" + joinTableName, showQuery);
                for (Object key : joinTableRecords.keySet())
                {
                    PropertyAccessor accessor = PropertyAccessorFactory.getPropertyAccessor((Field) entityMetadata
                            .getIdAttribute().getJavaMember());
                    byte[] rowKey = accessor.toBytes(key);

                    Set<Object> values = joinTableRecords.get(key);
                    List<Column> columns = new ArrayList<Column>();

                    // Create Insertion List
                    List<Mutation> insertionList = new ArrayList<Mutation>();
                    Class columnType = null;
                    for (Object value : values)
                    {
                        Column column = new Column();
                        column.setName(PropertyAccessorFactory.STRING.toBytes(invJoinColumnName
                                + Constants.JOIN_COLUMN_NAME_SEPARATOR + value));
                        column.setValue(PropertyAccessorHelper.getBytes(value));

                        column.setTimestamp(generator.getTimestamp());
                        columnType = value.getClass();
                        columns.add(column);

                        Mutation mut = new Mutation();
                        mut.setColumn_or_supercolumn(new ColumnOrSuperColumn().setColumn(column));
                        insertionList.add(mut);
                    }
                    createIndexesOnColumns(entityMetadata, joinTableName, columns, columnType);
                    // Create Mutation Map
                    Map<String, List<Mutation>> columnFamilyValues = new HashMap<String, List<Mutation>>();
                    columnFamilyValues.put(joinTableName, insertionList);
                    Map<ByteBuffer, Map<String, List<Mutation>>> mulationMap = new HashMap<ByteBuffer, Map<String, List<Mutation>>>();
                    mulationMap.put(ByteBuffer.wrap(rowKey), columnFamilyValues);

                    // Write Mutation map to database

                    conn.getClient().set_keyspace(entityMetadata.getSchema());

                    conn.getClient().batch_mutate(mulationMap, getConsistencyLevel());
                }
            }
        }
        catch (InvalidRequestException e)
        {
            log.error("Error while inserting record into join table, Caused by: .", e);
            throw new PersistenceException(e);
        }
        catch (TException e)
        {
            log.error("Error while inserting record into join table, Caused by: .", e);
            throw new PersistenceException(e);
        }
        finally
        {
            releaseConnection(conn);
        }
    }
@ Override public void persistJoinTable ( JoinTableData joinTableData ) { String joinTableName = joinTableData . getJoinTableName ( ) ; String invJoinColumnName = joinTableData . getInverseJoinColumnName ( ) ; Map < Object , Set < Object > > joinTableRecords = joinTableData . getJoinTableRecords ( ) ; EntityMetadata entityMetadata = KunderaMetadataManager . getEntityMetadata ( kunderaMetadata , joinTableData . getEntityClass ( ) ) ; Connection conn = null ; try { conn = getConnection ( ) ; if ( isCql3Enabled ( entityMetadata ) ) { persistJoinTableByCql ( joinTableData , conn . getClient ( ) ) ; } else { KunderaCoreUtils . printQuery ( "Persist join table:" + joinTableName , showQuery ) ; for ( Object key : joinTableRecords . keySet ( ) ) { PropertyAccessor accessor = PropertyAccessorFactory . getPropertyAccessor ( ( Field ) entityMetadata . getIdAttribute ( ) . getJavaMember ( ) ) ; byte [ ] rowKey = accessor . toBytes ( key ) ; Set < Object > values = joinTableRecords . get ( key ) ; List < Column > columns = new ArrayList < Column > ( ) ; List < Mutation > insertionList = new ArrayList < Mutation > ( ) ; Class columnType = null ; for ( Object value : values ) { Column column = new Column ( ) ; column . setName ( PropertyAccessorFactory . STRING . toBytes ( invJoinColumnName + Constants . JOIN_COLUMN_NAME_SEPARATOR + value ) ) ; column . setValue ( PropertyAccessorHelper . getBytes ( value ) ) ; column . setTimestamp ( generator . getTimestamp ( ) ) ; columnType = value . getClass ( ) ; columns . add ( column ) ; Mutation mut = new Mutation ( ) ; mut . setColumn_or_supercolumn ( new ColumnOrSuperColumn ( ) . setColumn ( column ) ) ; insertionList . add ( mut ) ; } createIndexesOnColumns ( entityMetadata , joinTableName , columns , columnType ) ; Map < String , List < Mutation > > columnFamilyValues = new HashMap < String , List < Mutation > > ( ) ; columnFamilyValues . put ( joinTableName , insertionList ) ; Map < ByteBuffer , Map < String , List < Mutation > > > mulationMap = new HashMap < ByteBuffer , Map < String , List < Mutation > > > ( ) ; mulationMap . put ( ByteBuffer . wrap ( rowKey ) , columnFamilyValues ) ; conn . getClient ( ) . set_keyspace ( entityMetadata . getSchema ( ) ) ; conn . getClient ( ) . batch_mutate ( mulationMap , getConsistencyLevel ( ) ) ; } } } catch ( InvalidRequestException e ) { log . error ( "Error while inserting record into join table, Caused by: ." , e ) ; throw new PersistenceException ( e ) ; } catch ( TException e ) { log . error ( "Error while inserting record into join table, Caused by: ." , e ) ; throw new PersistenceException ( e ) ; } finally { releaseConnection ( conn ) ; } }
Persists a Join table record set into database.

@param joinTableData
the join table data