code: 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());
      }
docs: Persists records into Join Table.
  
  @param joinTableData
  the join table data