ARTEMIS-3285 potential duplicate messages with LVQ + non-destructive #3569
Conversation
|
|
||
| @Test | ||
| public void testMultipleLastValuesCore() throws Exception { | ||
| testMultipleLastValues(CoreConnection); |
There was a problem hiding this comment.
Can you instead use CFUtil.createConnectionFactory("AMQP", "tcp://localhost:61616")
or
CFUtil.createConnectionFactory("CORE", "tcp://localhost:61616")
We should probably remove the Connection supplier. It's a bit simpler to read.
There was a problem hiding this comment.
I used the ConnectionSupplier because that's what all the other tests in the class use and I find that consistency helps readability. I wouldn't necessarily mind removing the ConnectionSupplier altogether, but it's used in around 80 other places so that change is beyond the scope of this PR. All things being equal I think the ConnectionSupplier is at least as readable as the alternative you have proposed.
There was a problem hiding this comment.
I would rather deprecate it. .and move to CFUtil everywhere.
| if (tm == null) { | ||
| break; | ||
| } | ||
| System.out.println("Received: " + tm.getStringProperty("data") + "; total: " + ++received); |
There was a problem hiding this comment.
logger.debug please
I have cleaned up the test suite recently to not use System.out.prinltn
| if (occurrences > 1 && !dups.containsValue(Integer.parseInt(s))) { | ||
| dups.put(s, occurrences); | ||
| } | ||
| System.out.print(s + ","); |
There was a problem hiding this comment.
Can you use log.info
| private static class HolderReference implements MessageReference { | ||
|
|
||
| private final SimpleString prop; | ||
|
|
||
| private volatile boolean delivered = false; | ||
| private volatile AtomicBoolean delivered = new AtomicBoolean(false); |
There was a problem hiding this comment.
Please use the AtomicLongFieldUpdater instead?
There was a problem hiding this comment.
As far as I can tell you can't use AtomicLongFieldUpdater on a boolean.
There was a problem hiding this comment.
Theres a way to do something, ill try dig it out.
| packetSize + | ||
| " from credits, available now is " + | ||
| availableCredits); | ||
| } | ||
| } |
There was a problem hiding this comment.
You don't need to ack the message if not?
This will create a big leakage on the journal
There was a problem hiding this comment.
You need to at least test what would happen on journal... try the issue out, and look at ./artemis print data if there are duplicates on the journal. I believe you would hit worsened issues here.
you need better tests where you would look at the result on the journal I am afraid.
| @@ -497,4 +497,8 @@ default void errorProcessing(Consumer consumer, Throwable t, MessageReference me | |||
| default QueueConfiguration getQueueConfiguration() { | |||
| return null; | |||
| } | |||
|
|
|||
| default boolean canBeDelivered(MessageReference ref) { | |||
There was a problem hiding this comment.
This is also breaking the contract on the Queue...
You have Queue::deliver already checking the message on handle(ref, consumer);
Whatever happens here needs to be treated inside handle.
Perhaps having a specialized version of handle would be a better choice.
having this canBeDelivered happened at the proceedDeliver can create a lot of other issues... like I'm predicting with leaving references and message on the journal and paging.
| } | ||
| Assert.fail("Duplicate messages received " + sb); | ||
| } | ||
|
|
There was a problem hiding this comment.
I added the following code here:
HashMap<Integer, AtomicInteger> map = countJournal(server.getConfiguration());
map.forEach((a, b) -> System.out.println("record type " + a + ", counts=" + b));That would be all you need to check on the journal...
Right now the output here shows exactly what I predicted:
record type 32, counts=100
record type 33, counts=97
record type 34, counts=3
record type 45, counts=100
You will need some assertion on those records. You need to ack the previous message that was already delivered.
If you can do that on handle(MessageReference), make it specialized on LastValueQueue would be better than the method you added on the consumer.
|
there are also a few failures added on this PR: org.apache.activemq.artemis.tests.integration.amqp.JMSNonDestructiveTest.testNonDestructiveCoreProducerCoreConsumer[persistenceEnabled=false, scanPeriod=100] |
|
I'll go back to the drawing board and rework this. |
No description provided.