def test_insertion_and_commitlog_behavior_after_reaching_cdc_total_space(self):
    '\n        Test that C* behaves correctly when CDC tables have consumed all the\n        space available to them. In particular: after writing\n        cdc_total_space_in_mb MB into CDC commitlogs:\n        - CDC writes are rejected\n        - non-CDC writes are accepted\n        - on flush, CDC commitlogs are copied to cdc_raw\n        - on flush, non-CDC commitlogs are not copied to cdc_raw\n        This is a lot of behavior to validate in one test, but we do so to\n        avoid running multiple tests that each write 1MB of data to fill\n        cdc_total_space_in_mb.\n        '
    ks_name = 'ks'
    full_cdc_table_info = TableInfo(ks_name=ks_name, table_name='full_cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'full_cdc_tab'), options={'cdc': 'true'})
    configuration_overrides = {'cdc_total_space_in_mb': 4}
    (node, session) = self.prepare(ks_name=ks_name, configuration_overrides=configuration_overrides)
    session.execute(full_cdc_table_info.create_stmt)
    non_cdc_table_info = TableInfo(ks_name=ks_name, table_name='non_cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'non_cdc_tab'))
    session.execute(non_cdc_table_info.create_stmt)
    empty_cdc_table_info = TableInfo(ks_name=ks_name, table_name='empty_cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'empty_cdc_tab'), options={'cdc': 'true'})
    session.execute(empty_cdc_table_info.create_stmt)
    logger.debug('flushing non-CDC commitlogs')
    node.flush()
    logger.debug('beginning data insert to fill CDC commitlogs')
    rows_loaded = _write_to_cdc_write_failure(session, full_cdc_table_info.insert_stmt)
    assert (0 < rows_loaded), 'No CDC rows inserted. This may happen when cdc_total_space_in_mb > commitlog_segment_size_in_mb'
    commitlog_dir = os.path.join(node.get_path(), 'commitlogs')
    commitlogs_size = size_of_files_in_dir(commitlog_dir)
    logger.debug('Commitlog dir ({d}) is {b}B'.format(d=commitlog_dir, b=commitlogs_size))
    try:
        session.execute(full_cdc_table_info.insert_stmt)
        raise Exception('WriteFailure expected')
    except WriteFailure:
        pass
    try:
        session.execute(empty_cdc_table_info.insert_stmt)
        raise Exception('WriteFailure expected')
    except WriteFailure:
        pass
    node.drain()
    session.cluster.shutdown()
    node.stop()
    node.start(wait_for_binary_proto=True)
    session = self.patient_cql_connection(node)
    pre_non_cdc_write_cdc_raw_segments = _get_cdc_raw_files(node.get_path())
    before_cdc_state = []
    if (self.cluster.version() >= '4.0'):
        node1_path = os.path.join(node.get_path(), 'cdc_raw')
        before_cdc_state = [ReplayData.load(node1_path, name) for name in os.listdir(node1_path) if name.endswith('_cdc.idx')]
    pre_non_cdc_write_segments = _get_commitlog_files(node.get_path())
    non_cdc_prepared_insert = session.prepare(non_cdc_table_info.insert_stmt)
    session.execute(non_cdc_prepared_insert, ())
    (start, time_limit) = (time.time(), 600)
    rate_limited_debug = get_rate_limited_function(logger.debug, 5)
    logger.debug('writing to non-cdc table')
    while (_get_commitlog_files(node.get_path()) <= pre_non_cdc_write_segments):
        elapsed = (time.time() - start)
        rate_limited_debug('  non-cdc load step has lasted {s:.2f}s'.format(s=elapsed))
        assert (elapsed <= time_limit), "It's been over a {s}s and we haven't written a new commitlog segment. Something is wrong.".format(s=time_limit)
        execute_concurrent(session, ((non_cdc_prepared_insert, ()) for _ in range(1000)), concurrency=500, raise_on_first_error=True)
    node.drain()
    session.cluster.shutdown()
    if (self.cluster.version() < '4.0'):
        assert (pre_non_cdc_write_cdc_raw_segments == _get_cdc_raw_files(node.get_path()))
    else:
        node2_path = os.path.join(node.get_path(), 'cdc_raw')
        after_cdc_state = [ReplayData.load(node2_path, name) for name in os.listdir(node2_path) if name.endswith('_cdc.idx')]
        found = True
        for idx in before_cdc_state:
            idx_found = False
            for idx_two in after_cdc_state:
                if compare_replay_data(idx, idx_two):
                    idx_found = True
            if (not idx_found):
                found = False
                break
        if (not found):
            self._fail_and_print_sets(before_cdc_state, after_cdc_state, 'Found CDC index in before not matched in after (non-CDC write test)')
        orphan_found = False
        for idx_two in after_cdc_state:
            index_found = False
            for idx in before_cdc_state:
                if compare_replay_data(idx_two, idx):
                    index_found = True
            if (not index_found):
                orphan_found = True
                break
        if orphan_found:
            self._fail_and_print_sets(before_cdc_state, after_cdc_state, 'Found orphaned index file in after CDC state not in former.')
Test that C* behaves correctly when CDC tables have consumed all the
space available to them. In particular: after writing
cdc_total_space_in_mb MB into CDC commitlogs:
- CDC writes are rejected
- non-CDC writes are accepted
- on flush, CDC commitlogs are copied to cdc_raw
- on flush, non-CDC commitlogs are not copied to cdc_raw
This is a lot of behavior to validate in one test, but we do so to
avoid running multiple tests that each write 1MB of data to fill
cdc_total_space_in_mb.
def test_insertion_and_commitlog_behavior_after_reaching_cdc_total_space(self):
    '\n        Test that C* behaves correctly when CDC tables have consumed all the\n        space available to them. In particular: after writing\n        cdc_total_space_in_mb MB into CDC commitlogs:\n        - CDC writes are rejected\n        - non-CDC writes are accepted\n        - on flush, CDC commitlogs are copied to cdc_raw\n        - on flush, non-CDC commitlogs are not copied to cdc_raw\n        This is a lot of behavior to validate in one test, but we do so to\n        avoid running multiple tests that each write 1MB of data to fill\n        cdc_total_space_in_mb.\n        '
    ks_name = 'ks'
    full_cdc_table_info = TableInfo(ks_name=ks_name, table_name='full_cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'full_cdc_tab'), options={'cdc': 'true'})
    configuration_overrides = {'cdc_total_space_in_mb': 4}
    (node, session) = self.prepare(ks_name=ks_name, configuration_overrides=configuration_overrides)
    session.execute(full_cdc_table_info.create_stmt)
    non_cdc_table_info = TableInfo(ks_name=ks_name, table_name='non_cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'non_cdc_tab'))
    session.execute(non_cdc_table_info.create_stmt)
    empty_cdc_table_info = TableInfo(ks_name=ks_name, table_name='empty_cdc_tab', column_spec=_16_uuid_column_spec, insert_stmt=_get_16_uuid_insert_stmt(ks_name, 'empty_cdc_tab'), options={'cdc': 'true'})
    session.execute(empty_cdc_table_info.create_stmt)
    logger.debug('flushing non-CDC commitlogs')
    node.flush()
    logger.debug('beginning data insert to fill CDC commitlogs')
    rows_loaded = _write_to_cdc_write_failure(session, full_cdc_table_info.insert_stmt)
    assert (0 < rows_loaded), 'No CDC rows inserted. This may happen when cdc_total_space_in_mb > commitlog_segment_size_in_mb'
    commitlog_dir = os.path.join(node.get_path(), 'commitlogs')
    commitlogs_size = size_of_files_in_dir(commitlog_dir)
    logger.debug('Commitlog dir ({d}) is {b}B'.format(d=commitlog_dir, b=commitlogs_size))
    try:
        session.execute(full_cdc_table_info.insert_stmt)
        raise Exception('WriteFailure expected')
    except WriteFailure:
        pass
    try:
        session.execute(empty_cdc_table_info.insert_stmt)
        raise Exception('WriteFailure expected')
    except WriteFailure:
        pass
    node.drain()
    session.cluster.shutdown()
    node.stop()
    node.start(wait_for_binary_proto=True)
    session = self.patient_cql_connection(node)
    pre_non_cdc_write_cdc_raw_segments = _get_cdc_raw_files(node.get_path())
    before_cdc_state = []
    if (self.cluster.version() >= '4.0'):
        node1_path = os.path.join(node.get_path(), 'cdc_raw')
        before_cdc_state = [ReplayData.load(node1_path, name) for name in os.listdir(node1_path) if name.endswith('_cdc.idx')]
    pre_non_cdc_write_segments = _get_commitlog_files(node.get_path())
    non_cdc_prepared_insert = session.prepare(non_cdc_table_info.insert_stmt)
    session.execute(non_cdc_prepared_insert, ())
    (start, time_limit) = (time.time(), 600)
    rate_limited_debug = get_rate_limited_function(logger.debug, 5)
    logger.debug('writing to non-cdc table')
    while (_get_commitlog_files(node.get_path()) <= pre_non_cdc_write_segments):
        elapsed = (time.time() - start)
        rate_limited_debug('  non-cdc load step has lasted {s:.2f}s'.format(s=elapsed))
        assert (elapsed <= time_limit), "It's been over a {s}s and we haven't written a new commitlog segment. Something is wrong.".format(s=time_limit)
        execute_concurrent(session, ((non_cdc_prepared_insert, ()) for _ in range(1000)), concurrency=500, raise_on_first_error=True)
    node.drain()
    session.cluster.shutdown()
    if (self.cluster.version() < '4.0'):
        assert (pre_non_cdc_write_cdc_raw_segments == _get_cdc_raw_files(node.get_path()))
    else:
        node2_path = os.path.join(node.get_path(), 'cdc_raw')
        after_cdc_state = [ReplayData.load(node2_path, name) for name in os.listdir(node2_path) if name.endswith('_cdc.idx')]
        found = True
        for idx in before_cdc_state:
            idx_found = False
            for idx_two in after_cdc_state:
                if compare_replay_data(idx, idx_two):
                    idx_found = True
            if (not idx_found):
                found = False
                break
        if (not found):
            self._fail_and_print_sets(before_cdc_state, after_cdc_state, 'Found CDC index in before not matched in after (non-CDC write test)')
        orphan_found = False
        for idx_two in after_cdc_state:
            index_found = False
            for idx in before_cdc_state:
                if compare_replay_data(idx_two, idx):
                    index_found = True
            if (not index_found):
                orphan_found = True
                break
        if orphan_found:
            self._fail_and_print_sets(before_cdc_state, after_cdc_state, 'Found orphaned index file in after CDC state not in former.')