IGNITE-11970 CQ buffers cleanup after acknowledge receiving implemented #7934
Conversation
|
|
||
| qry.setRemoteTransformerFactory(factory); | ||
|
|
||
| qry.setLocalListener((evts) -> evts.forEach(e -> System.out.println("val=" + e))); |
There was a problem hiding this comment.
Should we eliminate System.out.println usage in test?
There was a problem hiding this comment.
Yes, it seems reasonable. I'll modify local listener and add additional check for updates count in the tests.
…es count added to ContinuousQueryBufferCleanupAbstractTest. codestyle fixed.
| for (int i = next; i < entries.length; i++) { | ||
| CacheContinuousQueryEntry e = entries[i]; | ||
|
|
||
| if (e != null && e.updateCounter() <= updateCntr) { |
There was a problem hiding this comment.
Do we need checking e for non-null value? I think we need to clean up entries that have update counter less than acked counter each time from the beginning of entries array.
We can have out of order events on backup due to cache messages reordering, that's why an entry which comes after the acked counter (this entry has less counter than acked counter from the client) will never be cleaned.
There was a problem hiding this comment.
Clean up from the beginning of entries array was implemented.
| if (e != null && e.updateCounter() <= updateCntr) { | ||
| entries[i] = null; | ||
|
|
||
| lastCleaned = i; |
There was a problem hiding this comment.
Due to my comment above we can remove lastCleaned position.
| /** | ||
| * @param updateCntr Acknowledged counter. | ||
| */ | ||
| synchronized void cleanupEntries(Long updateCntr) { |
There was a problem hiding this comment.
Let's use long primitive type here to avoid unboxing each time comparing update counters.
| sendBackupAcknowledge(ackBuf.onAcknowledged(batch), routineId, ctx); | ||
| sendBackupAcknowledge(ackBufBackup.onAcknowledged(batch), routineId, ctx); | ||
|
|
||
| cleanupBuffers(ackBuf.onAcknowledged(batch)); |
There was a problem hiding this comment.
Can we achieve the same condition by calling lsnr.cleanupBackupQueue here with an acknowledged collection of counters? Why do we need a separate buffer for clearing partition buffers?
ackedTup = ackBufBackup.onAcknowledged(batch);
sendBackupAcknowledge(ackedTup, routineId, ctx);
lsnr.cleanupBackupQueue(ackedTup.get1());
There was a problem hiding this comment.
The separate buffer for backups is not required and was removed.
| } | ||
|
|
||
| @Override public void cleanupBackupQueue(Map<Integer, Long> updateCntrs) { | ||
| for (Map.Entry<Integer, Long> e : updateCntrs.entrySet()) { |
There was a problem hiding this comment.
It seems the cleanupBackupQueue becomes not the right name for this case. Let's rename it to something like cleanupPartitionBuffers.
There was a problem hiding this comment.
Renamed to "cleanupOnAck"
| srvs[i] = startGrid("srv" + i); | ||
|
|
||
| Ignite qryNode = useClient | ||
| ? startGrid(optimize(getConfiguration("client").setClientMode(true))) |
There was a problem hiding this comment.
Let's use startClientGrid().
There was a problem hiding this comment.
startClientGrid() used
| private void checkBuffer(int srvCnt, int backupsCnt, boolean useClient) throws Exception { | ||
| System.setProperty("IGNITE_CONTINUOUS_QUERY_ACK_THRESHOLD", Integer.toString(ACK_THRESHOLD)); | ||
|
|
||
| IgniteEx[] srvs = new IgniteEx[srvCnt]; |
There was a problem hiding this comment.
It seems this array is not needed. Let's use grid(0) instead of srv[0].
| for (int i = 0; i < RECORDS_CNT; i++) | ||
| cache.put(i, Integer.toString(i)); | ||
|
|
||
| Thread.sleep(2000); |
There was a problem hiding this comment.
Why do we need sleep here? I think we should use the waitForCondition here that all of the entries are processed by CQ.
There was a problem hiding this comment.
waitForCondition used instead of sleep.
| private static final int RECORDS_CNT = 10000; | ||
|
|
||
| /** */ | ||
| private static final int ACK_THRESHOLD = 100; |
There was a problem hiding this comment.
I think the constant from the CacheContinuousQueryHandler should be used here.
|
|
||
| IgniteCache<Integer, String> cache = qryNode.getOrCreateCache(cacheCfg); | ||
|
|
||
| AbstractContinuousQuery<Integer, String> qry = getContinuousQuery(); |
There was a problem hiding this comment.
It seems the @Parameterized as a continuous query can also be used here. So we can reduce the number of tests added to IgniteCacheQuerySelfTestSuite6. Can you, please, take a look?
# Conflicts: # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEventBuffer.java # modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java # modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite6.java
…anup execution changed. Parameterized tests implemented.
lobanovdmitry
left a comment
lobanovdmitry
left a comment
There was a problem hiding this comment.
Hello, trying to test this patch, found that one small change in
src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java still needed.
lines 551-553
@Override public void acknowledgeBackupOnTimeout(GridKernalContext ctx) {
sendBackupAcknowledge(ackBuf.acknowledgeOnTimeout(), routineId, ctx);
}
This callback is called by BackupCleaner and in some circumstances it releases AcknowledgeBuffer without calling
cleanupBuffers(acknowledged);
Please, consider this.
And thank you very much for the changes.
3 participants
Thank you for submitting the pull request to the Apache Ignite.
In order to streamline the review of the contribution
we ask you to ensure the following steps have been taken:
The Contribution Checklist
The description explains WHAT and WHY was made instead of HOW.
The following pattern must be used:
IGNITE-XXXX Change summarywhereXXXX- number of JIRA issue.(see the Maintainers list)
the
green visaattached to the JIRA ticket (see TC.Bot: Check PR)Notes
If you need any help, please email dev@ignite.apache.org or ask anу advice on http://asf.slack.com #ignite channel.