Skip to content

Commit e340b75

Browse files
CaideyipiCaideyipi
committed
fix
1 parent 2c62c39 commit e340b75

4 files changed

Lines changed: 242 additions & 74 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/UDFInfo.java‎

Lines changed: 13 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2424
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
2525
import org.apache.iotdb.commons.executable.ExecutableManager;
26+
import org.apache.iotdb.commons.executable.ReferenceCountedJarMetaKeeper;
2627
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
2728
import org.apache.iotdb.commons.udf.UDFInformation;
2829
import org.apache.iotdb.commons.udf.UDFTable;
@@ -41,7 +42,6 @@
4142
import org.apache.iotdb.rpc.TSStatusCode;
4243
import org.apache.iotdb.udf.api.exception.UDFManagementException;
4344

44-
import org.apache.tsfile.utils.ReadWriteIOUtils;
4545
import org.slf4j.Logger;
4646
import org.slf4j.LoggerFactory;
4747

@@ -54,7 +54,6 @@
5454
import java.nio.ByteBuffer;
5555
import java.util.ArrayList;
5656
import java.util.Collections;
57-
import java.util.HashMap;
5857
import java.util.List;
5958
import java.util.Map;
6059
import java.util.concurrent.locks.ReentrantLock;
@@ -67,8 +66,7 @@ public class UDFInfo implements SnapshotProcessor {
6766
ConfigNodeDescriptor.getInstance().getConf();
6867

6968
private final UDFTable udfTable;
70-
private final Map<String, String> existedJarToMD5;
71-
private final Map<String, Integer> existedJarToReferenceCount;
69+
private final ReferenceCountedJarMetaKeeper jarMetaKeeper;
7270

7371
private final UDFExecutableManager udfExecutableManager;
7472

@@ -78,8 +76,7 @@ public class UDFInfo implements SnapshotProcessor {
7876

7977
public UDFInfo() throws IOException {
8078
udfTable = new UDFTable();
81-
existedJarToMD5 = new HashMap<>();
82-
existedJarToReferenceCount = new HashMap<>();
79+
jarMetaKeeper = new ReferenceCountedJarMetaKeeper();
8380
udfExecutableManager =
8481
UDFExecutableManager.setupAndGetInstance(
8582
CONFIG_NODE_CONF.getUdfTemporaryLibDir(), CONFIG_NODE_CONF.getUdfDir());
@@ -106,7 +103,8 @@ public void validate(Model model, String udfName, String jarName, String jarMD5)
106103
TSStatusCode.UDF_ALREADY_EXISTS.getStatusCode());
107104
}
108105

109-
if (existedJarToMD5.containsKey(jarName) && !existedJarToMD5.get(jarName).equals(jarMD5)) {
106+
if (jarMetaKeeper.containsJar(jarName)
107+
&& !jarMetaKeeper.jarNameExistsAndMatchesMd5(jarName, jarMD5)) {
110108
throw new IoTDBRuntimeException(
111109
String.format(
112110
ConfigNodeMessages.FAILED_TO_CREATE_UDF_THE_SAME_NAME_JAR_BUT_DIFFERENT,
@@ -129,7 +127,7 @@ public UDFInformation getUDFInformation(Model model, String udfName)
129127
}
130128

131129
public boolean needToSaveJar(String jarName) {
132-
return !existedJarToMD5.containsKey(jarName);
130+
return jarMetaKeeper.needToSaveJar(jarName);
133131
}
134132

135133
public TSStatus addUDFInTable(CreateFunctionPlan physicalPlan) {
@@ -209,7 +207,7 @@ public Map<Model, Map<String, UDFInformation>> getRawUDFTable() {
209207

210208
@TestOnly
211209
public Map<String, String> getRawExistedJarToMD5() {
212-
return existedJarToMD5;
210+
return jarMetaKeeper.getJarNameToMd5Map();
213211
}
214212

215213
@Override
@@ -260,46 +258,28 @@ public void processLoadSnapshot(File snapshotDir) throws IOException {
260258
}
261259

262260
public void serializeExistedJarToMD5(OutputStream outputStream) throws IOException {
263-
ReadWriteIOUtils.write(existedJarToMD5.size(), outputStream);
264-
for (Map.Entry<String, String> entry : existedJarToMD5.entrySet()) {
265-
ReadWriteIOUtils.write(entry.getKey(), outputStream);
266-
ReadWriteIOUtils.write(entry.getValue(), outputStream);
267-
}
261+
jarMetaKeeper.serializeJarNameToMd5(outputStream);
268262
}
269263

270264
public void deserializeExistedJarToMD5(InputStream inputStream) throws IOException {
271-
int size = ReadWriteIOUtils.readInt(inputStream);
272-
while (size > 0) {
273-
existedJarToMD5.put(
274-
ReadWriteIOUtils.readString(inputStream), ReadWriteIOUtils.readString(inputStream));
275-
size--;
276-
}
265+
jarMetaKeeper.deserializeJarNameToMd5(inputStream);
277266
}
278267

279268
public void clear() {
280-
existedJarToMD5.clear();
281-
existedJarToReferenceCount.clear();
269+
jarMetaKeeper.clear();
282270
udfTable.clear();
283271
}
284272

285273
private void addJarReference(String jarName, String jarMD5) {
286-
existedJarToMD5.putIfAbsent(jarName, jarMD5);
287-
existedJarToReferenceCount.merge(jarName, 1, Integer::sum);
274+
jarMetaKeeper.addReference(jarName, jarMD5);
288275
}
289276

290277
private void removeJarReference(String jarName) {
291-
final Integer referenceCount = existedJarToReferenceCount.get(jarName);
292-
if (referenceCount == null || referenceCount <= 1) {
293-
existedJarToReferenceCount.remove(jarName);
294-
existedJarToMD5.remove(jarName);
295-
return;
296-
}
297-
existedJarToReferenceCount.put(jarName, referenceCount - 1);
278+
jarMetaKeeper.removeReference(jarName);
298279
}
299280

300281
private void rebuildJarMetadataFromUDFTable() {
301-
existedJarToMD5.clear();
302-
existedJarToReferenceCount.clear();
282+
jarMetaKeeper.clear();
303283
for (UDFInformation udfInformation : udfTable.getAllInformationList()) {
304284
if (udfInformation.isUsingURI()) {
305285
addJarReference(udfInformation.getJarName(), udfInformation.getJarMD5());

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/executable/ReferenceCountedJarMetaKeeper.java‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.commons.executable;
21+
22+
import org.apache.tsfile.utils.ReadWriteIOUtils;
23+
24+
import java.io.IOException;
25+
import java.io.InputStream;
26+
import java.io.OutputStream;
27+
import java.util.HashMap;
28+
import java.util.Map;
29+
import java.util.Objects;
30+
31+
public class ReferenceCountedJarMetaKeeper {
32+
33+
private final Map<String, String> jarNameToMd5Map = new HashMap<>();
34+
private final Map<String, Integer> jarNameToReferenceCountMap = new HashMap<>();
35+
36+
public synchronized boolean containsJar(final String jarName) {
37+
return jarNameToMd5Map.containsKey(jarName);
38+
}
39+
40+
public synchronized boolean needToSaveJar(final String jarName) {
41+
return !containsJar(jarName);
42+
}
43+
44+
public synchronized boolean jarNameExistsAndMatchesMd5(final String jarName, final String md5) {
45+
return containsJar(jarName) && Objects.equals(jarNameToMd5Map.get(jarName), md5);
46+
}
47+
48+
public synchronized void addReference(final String jarName, final String md5) {
49+
if (jarNameToReferenceCountMap.containsKey(jarName)) {
50+
jarNameToReferenceCountMap.put(jarName, jarNameToReferenceCountMap.get(jarName) + 1);
51+
return;
52+
}
53+
54+
jarNameToReferenceCountMap.put(jarName, 1);
55+
jarNameToMd5Map.put(jarName, md5);
56+
}
57+
58+
public synchronized void removeReference(final String jarName) {
59+
final Integer referenceCount = jarNameToReferenceCountMap.get(jarName);
60+
if (referenceCount == null || referenceCount <= 1) {
61+
jarNameToReferenceCountMap.remove(jarName);
62+
jarNameToMd5Map.remove(jarName);
63+
return;
64+
}
65+
66+
jarNameToReferenceCountMap.put(jarName, referenceCount - 1);
67+
}
68+
69+
public synchronized void clear() {
70+
jarNameToMd5Map.clear();
71+
jarNameToReferenceCountMap.clear();
72+
}
73+
74+
public synchronized Map<String, String> getJarNameToMd5Map() {
75+
return new HashMap<>(jarNameToMd5Map);
76+
}
77+
78+
public synchronized void serializeJarNameToMd5(final OutputStream outputStream)
79+
throws IOException {
80+
ReadWriteIOUtils.write(jarNameToMd5Map.size(), outputStream);
81+
for (final Map.Entry<String, String> entry : jarNameToMd5Map.entrySet()) {
82+
ReadWriteIOUtils.write(entry.getKey(), outputStream);
83+
ReadWriteIOUtils.write(entry.getValue(), outputStream);
84+
}
85+
}
86+
87+
public synchronized void deserializeJarNameToMd5(final InputStream inputStream)
88+
throws IOException {
89+
clear();
90+
91+
int size = ReadWriteIOUtils.readInt(inputStream);
92+
while (size > 0) {
93+
addReference(
94+
ReadWriteIOUtils.readString(inputStream), ReadWriteIOUtils.readString(inputStream));
95+
size--;
96+
}
97+
}
98+
99+
public synchronized void serializeJarNameToMd5AndReferenceCount(final OutputStream outputStream)
100+
throws IOException {
101+
ReadWriteIOUtils.write(jarNameToMd5Map.size(), outputStream);
102+
for (final Map.Entry<String, String> entry : jarNameToMd5Map.entrySet()) {
103+
final String jarName = entry.getKey();
104+
ReadWriteIOUtils.write(jarName, outputStream);
105+
ReadWriteIOUtils.write(entry.getValue(), outputStream);
106+
ReadWriteIOUtils.write(jarNameToReferenceCountMap.getOrDefault(jarName, 0), outputStream);
107+
}
108+
}
109+
110+
public synchronized void deserializeJarNameToMd5AndReferenceCount(final InputStream inputStream)
111+
throws IOException {
112+
clear();
113+
114+
final int jarSize = ReadWriteIOUtils.readInt(inputStream);
115+
for (int i = 0; i < jarSize; i++) {
116+
final String jarName = ReadWriteIOUtils.readString(inputStream);
117+
jarNameToMd5Map.put(jarName, ReadWriteIOUtils.readString(inputStream));
118+
jarNameToReferenceCountMap.put(jarName, ReadWriteIOUtils.readInt(inputStream));
119+
}
120+
}
121+
}

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/meta/ConfigNodePipePluginMetaKeeper.java‎

Lines changed: 9 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -19,80 +19,48 @@
1919

2020
package org.apache.iotdb.commons.pipe.agent.plugin.meta;
2121

22-
import org.apache.tsfile.utils.ReadWriteIOUtils;
22+
import org.apache.iotdb.commons.executable.ReferenceCountedJarMetaKeeper;
2323

2424
import java.io.IOException;
2525
import java.io.InputStream;
2626
import java.io.OutputStream;
27-
import java.util.HashMap;
28-
import java.util.Map;
2927

3028
public class ConfigNodePipePluginMetaKeeper extends PipePluginMetaKeeper {
3129

32-
protected final Map<String, String> jarNameToMd5Map;
33-
protected final Map<String, Integer> jarNameToReferenceCountMap;
30+
protected final ReferenceCountedJarMetaKeeper jarMetaKeeper;
3431

3532
public ConfigNodePipePluginMetaKeeper() {
3633
super();
3734

38-
jarNameToMd5Map = new HashMap<>();
39-
jarNameToReferenceCountMap = new HashMap<>();
35+
jarMetaKeeper = new ReferenceCountedJarMetaKeeper();
4036
}
4137

4238
public synchronized boolean containsJar(String jarName) {
43-
return jarNameToMd5Map.containsKey(jarName);
39+
return jarMetaKeeper.containsJar(jarName);
4440
}
4541

4642
public synchronized boolean jarNameExistsAndMatchesMd5(String jarName, String md5) {
47-
return jarNameToMd5Map.containsKey(jarName) && jarNameToMd5Map.get(jarName).equals(md5);
43+
return jarMetaKeeper.jarNameExistsAndMatchesMd5(jarName, md5);
4844
}
4945

5046
public synchronized void addJarNameAndMd5(String jarName, String md5) {
51-
if (jarNameToReferenceCountMap.containsKey(jarName)) {
52-
jarNameToReferenceCountMap.put(jarName, jarNameToReferenceCountMap.get(jarName) + 1);
53-
} else {
54-
jarNameToReferenceCountMap.put(jarName, 1);
55-
jarNameToMd5Map.put(jarName, md5);
56-
}
47+
jarMetaKeeper.addReference(jarName, md5);
5748
}
5849

5950
public synchronized void removeJarNameAndMd5IfPossible(String jarName) {
60-
if (jarNameToReferenceCountMap.containsKey(jarName)) {
61-
int count = jarNameToReferenceCountMap.get(jarName);
62-
if (count == 1) {
63-
jarNameToReferenceCountMap.remove(jarName);
64-
jarNameToMd5Map.remove(jarName);
65-
} else {
66-
jarNameToReferenceCountMap.put(jarName, count - 1);
67-
}
68-
}
51+
jarMetaKeeper.removeReference(jarName);
6952
}
7053

7154
@Override
7255
public void processTakeSnapshot(OutputStream outputStream) throws IOException {
73-
ReadWriteIOUtils.write(jarNameToMd5Map.size(), outputStream);
74-
for (Map.Entry<String, String> entry : jarNameToMd5Map.entrySet()) {
75-
ReadWriteIOUtils.write(entry.getKey(), outputStream);
76-
ReadWriteIOUtils.write(entry.getValue(), outputStream);
77-
ReadWriteIOUtils.write(jarNameToReferenceCountMap.get(entry.getKey()), outputStream);
78-
}
56+
jarMetaKeeper.serializeJarNameToMd5AndReferenceCount(outputStream);
7957

8058
super.processTakeSnapshot(outputStream);
8159
}
8260

8361
@Override
8462
public void processLoadSnapshot(InputStream inputStream) throws IOException {
85-
jarNameToMd5Map.clear();
86-
jarNameToReferenceCountMap.clear();
87-
88-
final int jarSize = ReadWriteIOUtils.readInt(inputStream);
89-
for (int i = 0; i < jarSize; i++) {
90-
final String jarName = ReadWriteIOUtils.readString(inputStream);
91-
final String md5 = ReadWriteIOUtils.readString(inputStream);
92-
final int count = ReadWriteIOUtils.readInt(inputStream);
93-
jarNameToMd5Map.put(jarName, md5);
94-
jarNameToReferenceCountMap.put(jarName, count);
95-
}
63+
jarMetaKeeper.deserializeJarNameToMd5AndReferenceCount(inputStream);
9664

9765
super.processLoadSnapshot(inputStream);
9866
}

0 commit comments

Comments
 (0)