[Issue 513] Correct apparent logic error in batchContainer's hasSpace() method

[bschofield]

Fixes issue #513.

This issue was previously partially addressed by @zzzming in PR #528.

Motivation

Following PR #528, the logic for the hasSpace() function, which is used to determine whether a batch container has enough space for a new message, reads:

func (bc *keyBasedBatchContainer) hasSpace(payload []byte) bool {
	msgSize := uint32(len(payload))
	return bc.numMessages+1 < bc.maxMessages || (bc.buffer.ReadableBytes()+msgSize) < uint32(bc.maxBatchSize)
}

Currently, then, the batch container will be considered to have space (be non-full) until both the max-messages and the max-size limits are hit.

I think that is incorrect. Nearby in the code, the batch container is considered to be full if either of the conditions are hit:

// IsFull check if the size in the current batch exceeds the maximum size allowed by the batch
func (bc *batchContainer) IsFull() bool {
	return bc.numMessages >= bc.maxMessages || bc.buffer.ReadableBytes() > uint32(bc.maxBatchSize)
}

Because the logical rule is !(A || B) = (!A) && (!B), the || in hasSpace() should be an &&.

Modifications

Change the || in hasSpace() to an &&, so that the container will be considered full if either limit is hit.

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

Added TestMaxBatchSize().

Documentation

• Does this pull request introduce a new feature? no

[bschofield]

The tests now seem to be failing on TestNacksTracker(). Is that a consequence of this change, or an unrelated issue?

[cckellogg]

The tests now seem to be failing on TestNacksTracker(). Is that a consequence of this change, or an unrelated issue?

It's probably unrelated since the tests passed for the other go versions. The concerning part is there may be a flaky TestNacksTracker() test now?

I started a retest

[bschofield]

The concerning part is there may be a flaky TestNacksTracker() test now?

I would guess that it's probably this part of newNackMockedConsumer():

// since the client ticks at an interval of delay / 3
// wait another interval to ensure we get all messages
time.Sleep(testNackDelay + 101*time.Millisecond)

Since the sleep above happens inside a new goroutine, and (*negativeAcksTracker).track() also runs inside a new goroutine, it's probably possible in rare cases for the goroutine scheduling to interfere with the intended sequence of actions. Also the documentation says that time.Ticker is allowed to drop ticks, which I suppose could be happening.

[bschofield]

I added TestMaxBatchSize() to producer_test.go, based on TestMaxMessageSize().

From the results of that test, it looks like the < in the size test should actually be <=, to accommodate the case where the new payload fits exactly in the container.

[zzzming]

We should use this

bc.numMessages+1 <= bc.maxMessages 

Because this will accommodate an individual message when batch is disabled. Currently this configuration fails at [unable to add message to batch]

pulsar.ProducerOptions{
		Topic:               topicName,
		DisableBatching:     true,
		BatchingMaxMessages: 1,
		BatchingMaxSize:     1,
	})

[bschofield]

Made the additional tweak as suggested by @zzzming, also cleaned up IsFull() (which is not actually used anywhere as far as I can see).

[wolfstudy]

LGTM +1