Add IT that ALTERs right before Reader fetches next page #134
Conversation
Bouncheck
force-pushed
the
mockdriver3reader
branch
3 times, most recently
from
72b758f to
4a05251
Compare
Bouncheck
changed the title
Bouncheck
force-pushed
the
mockdriver3reader
branch
from
4a05251 to
69f392a
Compare
scylla-cdc-driver3/src/test/java/com/scylladb/cdc/cql/driver3/MockDriver3WorkerCQL.java
Outdated
Show resolved
Hide resolved
Bouncheck
force-pushed
the
mockdriver3reader
branch
2 times, most recently
from
d33847e to
390fe1d
Compare
|
I've rebased on top of master, which should now have changes that recreate schema. |
Bouncheck
force-pushed
the
mockdriver3reader
branch
from
925119f to
59c5e5d
Compare
Maven cache only for dependencies, it should not cache builds |
|
Found one bug in the reworked test yesterday, but some other bug still remains. |
Please include fixes into this PR, update it's name and description to reflect that. |
Bouncheck
force-pushed
the
mockdriver3reader
branch
from
59c5e5d to
e43072b
Compare
Setting this option is required for MockDriver3WorkerCQL to work properly. Otherwise the types won't match.
Exposes in CDCConsumer already present option in CQLConfiguration Builder for setting fetch size.
This will be needed for scylla-cdc-lib to have access to mock test class from driver3 module.
Necessary to access mock test classes.
In order to complete the task I need to make Driver3WorkerCQL not final. Additionally the Driver3Reader is made non-final and access is changed to protected. This is in order to mock it in a test. `findNext` method, `ResultSet rs`, `lastChangeId` are now protected instead of private too.
Adds the mock version of Driver3WorkerCQL. This mock is created with singular purpose of using it in AlterTableIT. The mock adds ReentrantReadWriteLocks that allow to block the readers in specific moments. In the case of AlterTableIT we want to block before fetching the next page. Reflection is used in order not to modify the actual code of Driver3MultiReader.
Adds requested test method that ALTERs the base table right before the next page is fetched by the internal Reader of the WorkerCQL used by CDCConsumer. This is achieved by waiting for read lock of ReentrantReadWriteLock until the changes safely propagate and the test method releases the write lock. Only after that the Reader is allowed to fetch the next page. The fetch size is intentionally set to low value of 1. The test data is limited to single primary key. This is in order to limit the number of queued concurrent tasks.
Bouncheck
force-pushed
the
mockdriver3reader
branch
from
e43072b to
6beb8c3
Compare
|
The initial difference came from my local cache, but after fixing that other bugs surfaced. All should be fixed now. |
|
|
||
| @Override | ||
| protected void findNext(CompletableFuture<Optional<RawChange>> fut) { | ||
| CompletableFuture.runAsync( () -> { |
There was a problem hiding this comment.
I looked closer at findNext, and now I don't think that we need to run this guy in separate feature.
Only place it tends to do something async is via Futures.addCallback, but it uses MoreExecutors.directExecutor(), so it is completely sync function, therefore we can and should dump CompletableFuture.runAsync( () -> { here
| Preconditions.checkNotNull(systemProperties.getProperty("scylla.docker.version")); | ||
|
|
||
| @Test | ||
| public void alterBaseTableAtRuntime() { |
There was a problem hiding this comment.
Can you either update this test to make use of nextRowLock, or create test that target both nextRowLock and nextPageLock
There was a problem hiding this comment.
What would be the point of the test using nextRowLock? I've added it but I don't understand it's purpose exactly. AFAIK already fetched rows will not react to the changes on the cluster so I'm not sure what is there to test.
There was a problem hiding this comment.
to be sure that it works fine when it happens
| verifySchemaBeforeAlter(firstChange.get()); | ||
| verifySchemaAfterAlter(firstChangeAfterFetchingNewPage.get()); | ||
|
|
||
| assertFalse(failedDueToInvalidTypeEx.get()); |
There was a problem hiding this comment.
I don't see you ever change failedDueToInvalidTypeEx
There was a problem hiding this comment.
this is a leftover, will remove that
| try { | ||
| // Access the LocalTransport field in CDCConsumer | ||
| java.lang.reflect.Field transportField = CDCConsumer.class.getDeclaredField("transport"); | ||
| transportField.setAccessible(true); | ||
| LocalTransport transport = (LocalTransport) transportField.get(consumer); | ||
|
|
||
| // Access the WorkerConfiguration.Builder field in LocalTransport | ||
| java.lang.reflect.Field builderField = LocalTransport.class.getDeclaredField("workerConfigurationBuilder"); | ||
| builderField.setAccessible(true); | ||
| WorkerConfiguration.Builder builder = (WorkerConfiguration.Builder) builderField.get(transport); | ||
|
|
||
| // Access the session field in CDCConsumer | ||
| java.lang.reflect.Field sessionField = CDCConsumer.class.getDeclaredField("session"); | ||
| sessionField.setAccessible(true); | ||
| Driver3Session driver3Session = (Driver3Session) sessionField.get(consumer); | ||
|
|
||
| // Create MockDriver3WorkerCQL and replace the cql field in the builder | ||
| MockDriver3WorkerCQL mockWorkerCQL = new MockDriver3WorkerCQL(driver3Session); | ||
| // Grab the lock to control it later on | ||
| nextPageLock = mockWorkerCQL.nextPageLock; | ||
| java.lang.reflect.Field cqlField = WorkerConfiguration.Builder.class.getDeclaredField("cql"); | ||
| cqlField.setAccessible(true); | ||
| cqlField.set(builder, mockWorkerCQL); | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("Failed to replace Driver3WorkerCQL with MockDriver3WorkerCQL", e); | ||
| } |
There was a problem hiding this comment.
Can you please find better way to do that, without hacking into instances.
If necessary add additional API to builders.
There was a problem hiding this comment.
It will require modifying the accesses in production code, but I can.
There was a problem hiding this comment.
Probably someone would want to override these classes, since most of it already has interfaces it will not be a big deal.
| + " KEY (column1, column2)) WITH cdc = {'enabled': 'true'};", | ||
| keyspace, table)); | ||
|
|
||
| AtomicInteger counter = new AtomicInteger(0); |
There was a problem hiding this comment.
| AtomicInteger counter = new AtomicInteger(0); | |
| AtomicInteger updatesCount = new AtomicInteger(0); |
| thread.start(); | ||
|
|
||
| AtomicBoolean failedDueToInvalidTypeEx = new AtomicBoolean(false); | ||
| AtomicBoolean unlockedNextPageLock = new AtomicBoolean(false); |
There was a problem hiding this comment.
| AtomicBoolean unlockedNextPageLock = new AtomicBoolean(false); | |
| AtomicBoolean alterTableIsDone = new AtomicBoolean(false); |
| AtomicBoolean consumedChangeAfterAlter = new AtomicBoolean(false); | ||
| AtomicLong lastConsumedTime = new AtomicLong(0); | ||
| AtomicReference<RawChange> firstChange, firstChangeAfterFetchingNewPage; | ||
| firstChange = new AtomicReference<>(null); |
There was a problem hiding this comment.
| firstChange = new AtomicReference<>(null); | |
| changeBeforeAlter = new AtomicReference<>(null); |
| AtomicLong lastConsumedTime = new AtomicLong(0); | ||
| AtomicReference<RawChange> firstChange, firstChangeAfterFetchingNewPage; | ||
| firstChange = new AtomicReference<>(null); | ||
| firstChangeAfterFetchingNewPage = new AtomicReference<>(null); |
There was a problem hiding this comment.
| firstChangeAfterFetchingNewPage = new AtomicReference<>(null); | |
| changeAfterAlter = new AtomicReference<>(null); |
| // and consumer has not consumed any change for the last 10 seconds | ||
| ReentrantReadWriteLock finalNextPageLock = nextPageLock; | ||
| Awaitility.await().atMost(60, TimeUnit.SECONDS).until(() -> finalNextPageLock.hasQueuedThreads() && (System.currentTimeMillis() - lastConsumedTime.get() > 10000)); | ||
| ResultSet rs = session.execute(String.format("ALTER TABLE %s.%s " + "ADD column4 int;", keyspace, table)); |
There was a problem hiding this comment.
Can we also have tests on:
- Updating UDT
- Removing column
- Changing column type (by removing and adding column with the same name, but different type)
It would be great if you can generalize the test
There was a problem hiding this comment.
I'll make the alter query a parameter, or if it will be more complex some kind of runnable/callable that performs the modification and use that as a test parameter.
2 participants
Adds requested test method that ALTERs the base table right before the next
page is fetched by the internal Reader of the WorkerCQL used by CDCConsumer.
This is achieved by waiting for read lock of ReentrantReadWriteLock until
the changes safely propagate and the test method releases the write lock.
Only after that the Reader is allowed to fetch the next page.
The fetch size is intentionally set to low value of 1.
The test data is limited to single primary key. This is in order to limit
the number of queued concurrent tasks.
Test-jar is now created to allow cdc-lib to use mocked class from driver3 module.
Addresses #133.