@Override
    @CheckReturnValue
    public boolean offer(Entry<? extends Data, ? extends Data> entry) {
        int partitionId = partitionService.getPartitionId(entry.getKey());
        int length = entry.getKey().totalSize() + entry.getValue().totalSize() - 2 * HeapData.TYPE_OFFSET;

        // if the entry is larger than usableChunkSize, send it in its own chunk. We avoid adding it to the
        // ByteArrayOutputStream since it would expand it beyond its maximum capacity.
        if (length > usableChunkCapacity) {
            return putAsyncToMap(partitionId, () -> {
                byte[] data = new byte[serializedByteArrayHeader.length + length + valueTerminator.length];
                totalKeys++;
                int offset = 0;
                System.arraycopy(serializedByteArrayHeader, 0, data, offset, serializedByteArrayHeader.length);
                offset += serializedByteArrayHeader.length - Bits.INT_SIZE_IN_BYTES;

                Bits.writeInt(data, offset, length + valueTerminator.length, useBigEndian);
                offset += Bits.INT_SIZE_IN_BYTES;

                copyWithoutHeader(entry.getKey(), data, offset);
                offset += entry.getKey().totalSize() - HeapData.TYPE_OFFSET;

                copyWithoutHeader(entry.getValue(), data, offset);
                offset += entry.getValue().totalSize() - HeapData.TYPE_OFFSET;

                System.arraycopy(valueTerminator, 0, data, offset, valueTerminator.length);

                return new HeapData(data);
            });
        }

        // if the buffer after adding this entry and terminator would exceed the capacity limit, flush it first
        CustomByteArrayOutputStream buffer = buffers[partitionId];
        if (buffer.size() + length + valueTerminator.length > buffer.capacityLimit && !flushPartition(partitionId)) {
            return false;
        }

        // append to buffer
        writeWithoutHeader(entry.getKey(), buffer);
        writeWithoutHeader(entry.getValue(), buffer);
        totalKeys++;
        return true;
    }
@Test
    public void when_cannotAutoFlush_then_offerReturnsFalse() {
        // When
        // artificially increase number of async ops so that the writer cannot proceed
        writer.numConcurrentAsyncOps.set(JetServiceBackend.MAX_PARALLEL_ASYNC_OPS);
        Entry<Data, Data> entry = entry(serialize("k"), serialize("v"));
        int entriesInChunk =
                (writer.usableChunkCapacity - writer.serializedByteArrayHeader.length) / serializedLength(entry);
        assertTrue("entriesInChunk=" + entriesInChunk, entriesInChunk > 1 && entriesInChunk < 10);
        for (int i = 0; i < entriesInChunk; i++) {
            assertTrue(writer.offer(entry));
        }

        // Then
        assertFalse("offer should not have succeeded, too many parallel operations", writer.offer(entry));

        writer.numConcurrentAsyncOps.set(0);
        assertTrue("offer should have succeeded", writer.offer(entry));
        assertTargetMapEntry("k", 0, serializedLength(entry) * entriesInChunk);
    }