jt2594838 commented on code in PR #17669:
URL: https://github.com/apache/iotdb/pull/17669#discussion_r3242365978
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -228,12 +231,32 @@ private void assignToSource(
});
}
- public void startAssignTo(final PipeRealtimeDataRegionSource source) {
+ public synchronized void startAssignTo(final PipeRealtimeDataRegionSource
source) {
matcher.register(source);
+ if (source.isNeedListenToTsFile()) {
+ listenToTsFileSourceCount++;
+ }
+ if (source.isNeedListenToInsertNode()) {
+ listenToInsertNodeSourceCount++;
+ }
}
- public void stopAssignTo(final PipeRealtimeDataRegionSource source) {
+ public synchronized void stopAssignTo(final PipeRealtimeDataRegionSource
source) {
matcher.deregister(source);
+ if (source.isNeedListenToTsFile()) {
+ listenToTsFileSourceCount--;
+ }
+ if (source.isNeedListenToInsertNode()) {
+ listenToInsertNodeSourceCount--;
+ }
+ }
+
+ public synchronized boolean shouldListenToTsFile() {
+ return listenToTsFileSourceCount > 0;
+ }
+
+ public synchronized boolean shouldListenToInsertNode() {
+ return listenToInsertNodeSourceCount > 0;
}
Review Comment:
Necessary to synchronize?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java:
##########
@@ -51,23 +50,18 @@ public class PipeInsertionDataNodeListener {
private final ConcurrentMap<Integer, PipeDataRegionAssigner>
dataRegionId2Assigner =
new ConcurrentHashMap<>();
- private final AtomicInteger listenToTsFileSourceCount = new AtomicInteger(0);
- private final AtomicInteger listenToInsertNodeSourceCount = new
AtomicInteger(0);
-
//////////////////////////// start & stop ////////////////////////////
public synchronized void startListenAndAssign(
final int dataRegionId, final PipeRealtimeDataRegionSource source) {
- dataRegionId2Assigner
- .computeIfAbsent(dataRegionId, o -> new
PipeDataRegionAssigner(dataRegionId))
- .startAssignTo(source);
-
- if (source.isNeedListenToTsFile()) {
- listenToTsFileSourceCount.incrementAndGet();
- }
- if (source.isNeedListenToInsertNode()) {
- listenToInsertNodeSourceCount.incrementAndGet();
- }
+ dataRegionId2Assigner.compute(
+ dataRegionId,
+ (id, assigner) -> {
+ final PipeDataRegionAssigner actualAssigner =
+ assigner == null ? new PipeDataRegionAssigner(dataRegionId) :
assigner;
+ actualAssigner.startAssignTo(source);
+ return actualAssigner;
+ });
}
Review Comment:
ComputeIfAbsent?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
