apache / apache/fluss

Fix OutOfBoundException in writeChangeType when the number of records is large

Open
#2,967 2 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
2.1k
Forks
625
Avg merge
3d 14h
Merged PRs (30d)
97

Description

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Fluss version

0.9.0 (latest release)

### Please describe the bug 🐞

In function `ChangeTypeVectorWriter.writeChangeType`, there is chance to see `OutOfBoundException` when a record batch is large.

> Suppressed: java.io.IOException: One or more Fluss Writer send requests have encountered exception
at org.apache.fluss.flink.sink.writer.FlinkSinkWriter.checkAsyncException(FlinkSinkWriter.java:239)
at org.apache.fluss.flink.sink.writer.FlinkSinkWriter.close(FlinkSinkWriter.java:198)
at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255)
at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:234)
at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:223)
at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.close(SinkWriterOperator.java:275)
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:163)
at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.closeAllOperators(RegularOperatorChain.java:125)
at org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:1410)
at org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255)
at org.apache.flink.core.fs.AutoCloseableRegistry.doClose(AutoCloseableRegistry.java:72)
at org.apache.flink.util.AbstractAutoCloseableRegistry.close(AbstractAutoCloseableRegistry.java:127)
at org.apache.flink.streaming.runtime.tasks.StreamTask.cleanUp(StreamTask.java:1314)
at org.apache.flink.runtime.taskmanager.Task.lambda$restoreAndInvoke$0(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:976)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:958)
... 3 more
Caused by: java.util.concurrent.CompletionException: org.apache.fluss.exception.UnknownServerException: java.lang.IndexOutOfBoundsException: capacity: 131072, index: 131072, put length: 1
at org.apache.fluss.memory.MemorySegment.put(MemorySegment.java:325)
at org.apache.fluss.record.ChangeTypeVectorWriter.writeChangeType(ChangeTypeVectorWriter.java:44)
at org.apache.fluss.record.MemoryLogRecordsArrowBuilder.append(MemoryLogRecordsArrowBuilder.java:185)
at org.apache.fluss.server.kv.wal.ArrowWalBuilder.append(ArrowWalBuilder.java:48)
at org.apache.fluss.server.kv.KvTablet.applyUpdate(KvTablet.java:624)
at org.apache.fluss.server.kv.KvTablet.processUpsert(KvTablet.java:582)
at org.apache.fluss.server.kv.KvTablet.processKvRecords(KvTablet.java:493)
at org.apache.fluss.server.kv.KvTablet.lambda$putAsLeader$0(KvTablet.java:406)
at org.apache.fluss.utils.concurrent.LockUtils.inLock(LockUtils.java:42)
at org.apache.fluss.utils.concurrent.LockUtils.inWriteLock(LockUtils.java:65)
at org.apache.fluss.server.kv.KvTablet.putAsLeader(KvTablet.java:367)
at org.apache.fluss.server.replica.Replica.lambda$putRecordsToLeader$9(Replica.java:1012)
at org.apache.fluss.utils.concurrent.LockUtils.inLock(LockUtils.java:42)
at org.apache.fluss.utils.concurrent.LockUtils.inReadLock(LockUtils.java:55)
at org.apache.fluss.server.replica.Replica.putRecordsToLeader(Replica.java:996)
at org.apache.fluss.server.replica.ReplicaManager.putToLocalKv(ReplicaManager.java:1282)
at org.apache.fluss.server.replica.ReplicaManager.putRecordsToKv(ReplicaManager.java:641)
at org.apache.fluss.server.tablet.TabletService.putKv(TabletService.java:238)
at jdk.internal.reflect.GeneratedMethodAccessor10.invoke(Unknown Source)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.apache.fluss.rpc.netty.server.FlussRequestHandler.processRequest(FlussRequestHandler.java:63)

### Solution

`ChangeTypeVectorWriter` only have one memory segment now. To solve this problem, we need to make the `ChangeTypeVectorWriter` support multiple memory pages.

And another minor problem is that the boundary check is not good.
> if (recordsCount > capacity) {
throw new IllegalStateException("The change type vector is full.");
}
segment.put(startPosition + recordsCount, changeType.toByteValue());

we should check `records >= capacity` and expect to see this `IllegalStateException`, rather than writing to memory and see a runtime `OutOfBoundException`

### Are you willing to submit a PR?

- [ ] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start with ChangeTypeVectorWriter.writeChangeType and trace how MemorySegment pages are supplied to the writer, using the stack trace through MemoryLogRecordsArrowBuilder.append as context. Update the writer to handle multiple memory pages and correct the recordsCount boundary check; done means large record batches no longer overflow and the expected IllegalStateException is used at capacity.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.