def test_cdc_data_available_in_cdc_raw(self):
    ks_name = 'ks'
    (generation_node, generation_session) = self.prepare(ks_name=ks_name)
    cdc_table_info = TableInfo(ks_name=ks_name, table_name='cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'cdc_tab'), options={'cdc': 'true', 'id': uuid.uuid4()})
    advance_to_next_cl_segment(session=generation_session, commitlog_dir=os.path.join(generation_node.get_path(), 'commitlogs'))
    generation_session.execute(cdc_table_info.create_stmt)
    inserted_rows = _insert_rows(generation_session, cdc_table_info.name, cdc_table_info.insert_stmt, repeat((), 10000))
    logger.debug('draining')
    generation_node.drain()
    logger.debug('stopping')
    generation_session.cluster.shutdown()
    generation_node.stop()
    source_path = os.path.join(generation_node.get_path(), 'cdc_raw')
    source_cdc_indexes = {ReplayData.load(source_path, name) for name in source_path if name.endswith('_cdc.idx')}
    assert (source_cdc_indexes != {})
    loading_node = self._init_new_loading_node(ks_name, cdc_table_info.create_stmt, (self.cluster.version() < '4'))
    logger.debug('moving cdc_raw and restarting node')
    _move_commitlog_segments(os.path.join(generation_node.get_path(), 'cdc_raw'), os.path.join(loading_node.get_path(), 'commitlogs'))
    loading_node.start(wait_for_binary_proto=120)
    logger.debug('node successfully started; waiting on log replay')
    loading_node.grep_log('Log replay complete')
    logger.debug('log replay complete')
    validation_session = self.patient_exclusive_cql_connection(loading_node)
    data_in_cdc_table_after_restart = rows_to_list(validation_session.execute(('SELECT * FROM ' + cdc_table_info.name)))
    logger.debug('found {cdc} values in CDC table'.format(cdc=len(data_in_cdc_table_after_restart)))
    assert (inserted_rows == data_in_cdc_table_after_restart), 'not all expected data selected'
    if (self.cluster.version() >= '4.0'):
        loading_path = os.path.join(loading_node.get_path(), 'cdc_raw')
        dest_cdc_indexes = [ReplayData.load(loading_path, name) for name in os.listdir(loading_path) if name.endswith('_cdc.idx')]
        for srd in source_cdc_indexes:
            assert os.path.isfile(os.path.join(loading_path, srd.idx_name))
            assert os.path.isfile(os.path.join(loading_path, srd.log_name))
            corresponding_dest_replay_datae = [x for x in dest_cdc_indexes if (srd.idx_name == x.idx_name)]
            assert_length_equal(corresponding_dest_replay_datae, 1)
            drd = corresponding_dest_replay_datae[0]
            assert (drd.offset >= srd.offset)
            assert (srd.completed == drd.completed)
        src_to_dest_idx_map = {src_rd: [dest_rd for dest_rd in dest_cdc_indexes if (dest_rd.idx_name == src_rd.idx_name)] for src_rd in source_cdc_indexes}
        for (src_rd, dest_rds) in src_to_dest_idx_map.items():
            assert_length_equal(dest_rds, 1)
            src_to_dest_idx_map[src_rd] = dest_rds[0]
        assert (0 not in {i.offset for i in src_to_dest_idx_map.values()}), 'Found index offsets == 0 in an index file on the destination node that corresponds to an index file on the source node:\n{}'.format(pformat(src_to_dest_idx_map))
        for (src_rd, dest_rd) in src_to_dest_idx_map.items():
            assert (dest_rd.offset >= src_rd.offset)
        src_to_dest_idx_map = {src_rd: [dest_rd for dest_rd in dest_cdc_indexes if (dest_rd.idx_name == src_rd.idx_name)] for src_rd in source_cdc_indexes}
        for (k, v) in src_to_dest_idx_map.items():
            assert_length_equal(v, 1)
            assert (k.offset >= v.offset)
def test_cdc_data_available_in_cdc_raw(self):
    ks_name = 'ks'
    (generation_node, generation_session) = self.prepare(ks_name=ks_name)
    cdc_table_info = TableInfo(ks_name=ks_name, table_name='cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'cdc_tab'), options={'cdc': 'true', 'id': uuid.uuid4()})
    advance_to_next_cl_segment(session=generation_session, commitlog_dir=os.path.join(generation_node.get_path(), 'commitlogs'))
    generation_session.execute(cdc_table_info.create_stmt)
    inserted_rows = _insert_rows(generation_session, cdc_table_info.name, cdc_table_info.insert_stmt, repeat((), 10000))
    logger.debug('draining')
    generation_node.drain()
    logger.debug('stopping')
    generation_session.cluster.shutdown()
    generation_node.stop()
    source_path = os.path.join(generation_node.get_path(), 'cdc_raw')
    source_cdc_indexes = {ReplayData.load(source_path, name) for name in source_path if name.endswith('_cdc.idx')}
    assert (source_cdc_indexes != {})
    loading_node = self._init_new_loading_node(ks_name, cdc_table_info.create_stmt, (self.cluster.version() < '4'))
    logger.debug('moving cdc_raw and restarting node')
    _move_commitlog_segments(os.path.join(generation_node.get_path(), 'cdc_raw'), os.path.join(loading_node.get_path(), 'commitlogs'))
    loading_node.start(wait_for_binary_proto=120)
    logger.debug('node successfully started; waiting on log replay')
    loading_node.grep_log('Log replay complete')
    logger.debug('log replay complete')
    validation_session = self.patient_exclusive_cql_connection(loading_node)
    data_in_cdc_table_after_restart = rows_to_list(validation_session.execute(('SELECT * FROM ' + cdc_table_info.name)))
    logger.debug('found {cdc} values in CDC table'.format(cdc=len(data_in_cdc_table_after_restart)))
    assert (inserted_rows == data_in_cdc_table_after_restart), 'not all expected data selected'
    if (self.cluster.version() >= '4.0'):
        loading_path = os.path.join(loading_node.get_path(), 'cdc_raw')
        dest_cdc_indexes = [ReplayData.load(loading_path, name) for name in os.listdir(loading_path) if name.endswith('_cdc.idx')]
        for srd in source_cdc_indexes:
            assert os.path.isfile(os.path.join(loading_path, srd.idx_name))
            assert os.path.isfile(os.path.join(loading_path, srd.log_name))
            corresponding_dest_replay_datae = [x for x in dest_cdc_indexes if (srd.idx_name == x.idx_name)]
            assert_length_equal(corresponding_dest_replay_datae, 1)
            drd = corresponding_dest_replay_datae[0]
            assert (drd.offset >= srd.offset)
            assert (srd.completed == drd.completed)
        src_to_dest_idx_map = {src_rd: [dest_rd for dest_rd in dest_cdc_indexes if (dest_rd.idx_name == src_rd.idx_name)] for src_rd in source_cdc_indexes}
        for (src_rd, dest_rds) in src_to_dest_idx_map.items():
            assert_length_equal(dest_rds, 1)
            src_to_dest_idx_map[src_rd] = dest_rds[0]
        assert (0 not in {i.offset for i in src_to_dest_idx_map.values()}), 'Found index offsets == 0 in an index file on the destination node that corresponds to an index file on the source node:\n{}'.format(pformat(src_to_dest_idx_map))
        for (src_rd, dest_rd) in src_to_dest_idx_map.items():
            assert (dest_rd.offset >= src_rd.offset)
        src_to_dest_idx_map = {src_rd: [dest_rd for dest_rd in dest_cdc_indexes if (dest_rd.idx_name == src_rd.idx_name)] for src_rd in source_cdc_indexes}
        for (k, v) in src_to_dest_idx_map.items():
            assert_length_equal(v, 1)
            assert (k.offset >= v.offset)