This is an automated email from the ASF dual-hosted git repository.
saihemanth-cloudera pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new f298b9ccee8 HIVE-29784: Cleaning lots of expired notifications could
lead to OOM (#6734)
f298b9ccee8 is described below
commit f298b9ccee8fe3715196aeeeaa46f81bccf77043
Author: dengzh <[email protected]>
AuthorDate: Tue Sep 29 01:37:54 2026 +0800
HIVE-29784: Cleaning lots of expired notifications could lead to OOM (#6734)
---
.../apache/hadoop/hive/metastore/ObjectStore.java | 11 +-
.../metastore/iface/NotificationStore.java | 6 +-
.../metastore/impl/ColStatsStoreImpl.java | 2 +-
.../metastore/impl/NotificationStoreImpl.java | 42 ++++---
.../metastore/metastore/impl/TableStoreImpl.java | 61 +++++-----
.../hive/metastore/ExecutionContextTestUtils.java | 57 ++++++++++
.../metastore/TestPersistenceContextEviction.java | 124 +++++++++++++++++++++
7 files changed, 248 insertions(+), 55 deletions(-)
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
index 3ca67b9f6b1..6e324ad4679 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java
@@ -421,12 +421,11 @@ public <T> T unwrap(Class<T> iface) {
if (descriptor == null) {
throw new IllegalArgumentException("Unable to unwrap the store as " +
iface);
}
- String implClassName =
- conf.get("metastore." + descriptor.alias() + ".store.impl", "");
- Class<?> ifaceImpl = descriptor.defaultImpl();
- if (StringUtils.isNotEmpty(implClassName)) {
- ifaceImpl = conf.getClass(implClassName, ifaceImpl);
- }
+ String ifaceKey = "metastore." + descriptor.alias() + ".store.impl";
+ String implClassName = conf.getTrimmed(ifaceKey);
+ Class<?> ifaceImpl = StringUtils.isNotEmpty(implClassName)
+ ? conf.getClass(ifaceKey, descriptor.defaultImpl())
+ : descriptor.defaultImpl();
T simpl = (T) JavaUtils.newInstance(ifaceImpl);
List<Query> openQueries = new LinkedList<>();
if (simpl instanceof RawStoreBundle rsb) {
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java
index ed0b743e109..9f7737594b7 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/iface/NotificationStore.java
@@ -50,10 +50,11 @@ public interface NotificationStore {
void addNotificationEvent(NotificationEvent event) throws MetaException;
/**
- * Remove older notification events.
+ * Remove older notification events, transaction is explicitly handled
inside.
*
* @param olderThan Remove any events older or equal to a given number of
seconds
*/
+ @MetaDescriptor.NoTransaction
void cleanNotificationEvents(int olderThan);
/**
@@ -72,9 +73,10 @@ public interface NotificationStore {
NotificationEventsCountResponse
getNotificationEventsCount(NotificationEventsCountRequest rqst);
/**
- * Remove older notification events.
+ * Remove older notification events, transaction is explicitly handled
inside.
* @param olderThan Remove any events older or equal to a given number of
seconds
*/
+ @MetaDescriptor.NoTransaction
void cleanWriteNotificationEvents(int olderThan);
/**
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java
index be1a672d354..fcfb3c9d5a1 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/ColStatsStoreImpl.java
@@ -484,7 +484,6 @@ private List<MTableColumnStatistics>
getMTableColumnStatistics(Table table, List
validateTableCols(table, colNames);
List<MTableColumnStatistics> result = Collections.emptyList();
- Query query = pm.newQuery(MTableColumnStatistics.class);
result =
Batchable.runBatched(batchSize, colNames, new Batchable<String,
MTableColumnStatistics>() {
@Override
@@ -505,6 +504,7 @@ public List<MTableColumnStatistics> run(List<String> input)
params[i + 4] = input.get(i);
}
filter.append(")");
+ Query query = pm.newQuery(MTableColumnStatistics.class);
query.setFilter(filter.toString());
query.declareParameters(paramStr.toString());
List<MTableColumnStatistics> paritial =
(List<MTableColumnStatistics>) query.executeWithArray(params);
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java
index 89c29abebaa..b0dd0d27d8d 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/NotificationStoreImpl.java
@@ -248,16 +248,15 @@ private void cleanOlderEvents(int olderThan, Class table,
String tableName) {
final Optional<Integer> batchSize = (eventBatchSize > 0) ?
Optional.of(eventBatchSize) : Optional.empty();
final long start = System.nanoTime();
- int deleteCount = doCleanNotificationEvents(tooOld, batchSize, table,
tableName);
+ int deleteCount = 0;
+ int batchCount;
+ do {
+ batchCount = cleanNotificationEventsBatch(tooOld, batchSize, table,
tableName);
+ deleteCount += batchCount;
+ } while (batchCount > 0);
if (deleteCount == 0) {
LOG.info("No {} events found to be cleaned with eventTime < {}",
tableName, tooOld);
- } else {
- int batchCount = 0;
- do {
- batchCount = doCleanNotificationEvents(tooOld, batchSize, table,
tableName);
- deleteCount += batchCount;
- } while (batchCount > 0);
}
final long finish = System.nanoTime();
@@ -265,6 +264,21 @@ private void cleanOlderEvents(int olderThan, Class table,
String tableName) {
TimeUnit.NANOSECONDS.toMillis(finish - start));
}
+ private <T> int cleanNotificationEventsBatch(final int ageSec, final
Optional<Integer> batchSize,
+ Class<T> tableClass, String tableName) {
+ boolean committed = false;
+ baseStore.openTransaction();
+ try {
+ int deleted = doCleanNotificationEvents(ageSec, batchSize, tableClass,
tableName);
+ committed = baseStore.commitTransaction();
+ return deleted;
+ } finally {
+ if (!committed && baseStore.isActiveTransaction()) {
+ baseStore.rollbackTransaction();
+ }
+ }
+ }
+
private <T> int doCleanNotificationEvents(final int ageSec, final
Optional<Integer> batchSize,
Class<T> tableClass, String tableName) {
int eventsCount = 0;
@@ -309,6 +323,7 @@ private <T> int doCleanNotificationEvents(final int ageSec,
final Optional<Integ
}
pm.deletePersistentAll(events);
}
+ query.closeAll();
return eventsCount;
}
@@ -328,15 +343,11 @@ public NotificationEventsCountResponse
getNotificationEventsCount(NotificationEv
Long result = 0L;
long fromEventId = rqst.getFromEventId();
String inputDbName = rqst.getDbName();
- String catName = rqst.isSetCatName() ? rqst.getCatName() :
getDefaultCatalog(conf);
+ String catName = rqst.isSetCatName() ?
normalizeIdentifier(rqst.getCatName()) : getDefaultCatalog(conf);
long toEventId;
String paramSpecs;
List<Object> paramVals = new ArrayList<>();
- // We store a catalog name in lower case in metastore and also use the
same way everywhere in
- // hive.
- assert catName.equals(catName.toLowerCase());
-
// Build the query to count events, part by part
String queryStr = "select count(eventId) from " +
MNotificationLog.class.getName();
// count fromEventId onwards events
@@ -353,8 +364,7 @@ public NotificationEventsCountResponse
getNotificationEventsCount(NotificationEv
// counted.
queryStr = queryStr + " && (dbName == inputDbName || dbName == null)";
paramSpecs = paramSpecs + ", java.lang.String inputDbName";
- // We store a database name in lower case in metastore.
- paramVals.add(inputDbName.toLowerCase());
+ paramVals.add(normalizeIdentifier(inputDbName));
}
// catName could be NULL in case of transaction related events, which also
need to be
@@ -374,7 +384,7 @@ public NotificationEventsCountResponse
getNotificationEventsCount(NotificationEv
if (rqst.isSetTableNames() && !rqst.getTableNames().isEmpty()) {
queryStr = queryStr + " && (";
for (String tableName : rqst.getTableNames()) {
- paramVals.add(tableName.toLowerCase());
+ paramVals.add(normalizeIdentifier(tableName));
queryStr = queryStr + "tableName == tableName" + paramVals.size() + "
|| ";
paramSpecs = paramSpecs + ", java.lang.String tableName" +
paramVals.size();
}
@@ -386,7 +396,7 @@ public NotificationEventsCountResponse
getNotificationEventsCount(NotificationEv
query.declareParameters(paramSpecs);
result = (Long) query.executeWithArray(paramVals.toArray());
// Cap the event count by limit if specified.
- long eventCount = result.longValue();
+ long eventCount = result.longValue();
if (rqst.isSetLimit() && eventCount > rqst.getLimit()) {
eventCount = rqst.getLimit();
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
index 26f76bc92f0..f1fa5aaac40 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
@@ -938,7 +938,7 @@ public Table alterTable(TableName tableName, Table
newTable, String queryValidWr
boolean isToTxn = isTxn &&
!TxnUtils.isTransactionalTable(oldt.getParameters());
if (!isToTxn && isTxn && areTxnStatsSupported) {
// Transactional table is altered without a txn. Make sure there are no
changes to the flag.
- String errorMsg = verifyStatsChangeCtx(TableName.getDbTable(name,
dbname), oldt.getParameters(),
+ String errorMsg = verifyStatsChangeCtx(TableName.getDbTable(dbname,
name), oldt.getParameters(),
newTable.getParameters(), newTable.getWriteId(), queryValidWriteIds,
false);
if (errorMsg != null) {
throw new MetaException(errorMsg);
@@ -1677,7 +1677,8 @@ protected List<Partition> getSqlResult() throws
MetaException {
// We couldn't do SQL filter pushdown. Get names via normal means.
List<String> partNames = new LinkedList<>();
hasUnknownPartitions.set(getPartitionNamesPrunedByExprNoTxn(
- catName, dbName, tblName, partitionKeys, expr,
args.getDefaultPartName(), (short) args.getMax(), partNames));
+ catName, dbName, tblName, partitionKeys, expr,
args.getDefaultPartName(),
+ (short) args.getMax(), partNames));
GetPartitionsArgs newArgs = new
GetPartitionsArgs.GetPartitionsArgsBuilder(args).partNames(partNames).build();
return getDirectSql().getPartitionsViaPartNames(catName, dbName,
tblName, newArgs);
}
@@ -1788,7 +1789,8 @@ public List<Partition> run(List<String> input) throws
MetaException {
* @param result The resulting names.
* @return Whether the result contains any unknown partitions.
*/
- private boolean getPartitionNamesPrunedByExprNoTxn(String catName, String
dbName, String tblName, List<FieldSchema> partColumns, byte[] expr,
+ private boolean getPartitionNamesPrunedByExprNoTxn(String catName, String
dbName, String tblName,
+ List<FieldSchema> partColumns, byte[] expr,
String defaultPartName, short maxParts, List<String> result) throws
MetaException {
result.addAll(getPartitionNamesNoTxn(catName, dbName, tblName, (short)
-1));
return prunePartitionNamesByExpr(catName, dbName, tblName, result,
@@ -1846,25 +1848,17 @@ public Partition alterPartition(TableName tableName,
List<String> part_vals, Par
String dbname = normalizeIdentifier(tableName.getDb());
String name = normalizeIdentifier(tableName.getTable());
AtomicReference<MColumnDescriptor> oldCd = new AtomicReference<>();
- Partition result = alterPartitionNoTxn(catName, dbname, name, part_vals,
new_part, queryValidWriteIds, oldCd);
+ MTable table = this.getMTable(new_part.getCatName(), new_part.getDbName(),
new_part.getTableName());
+ MPartition oldp = getMPartition(catName, dbname, name, part_vals, table);
+ Partition result = alterPartitionNoTxn(catName, dbname, name, oldp,
new_part, queryValidWriteIds, oldCd, table);
removeUnusedColumnDescriptor(oldCd.get());
return result;
}
/**
* Alters an existing partition. Initiates copy of SD. Returns the old CD.
- * @param part_vals Partition values (of the original partition instance)
* @param newPart Partition object containing new information
*/
- private Partition alterPartitionNoTxn(String catName, String dbname, String
name,
- List<String> part_vals, Partition newPart, String validWriteIds,
AtomicReference<MColumnDescriptor> oldCd)
- throws InvalidObjectException, MetaException {
- MTable table = this.getMTable(newPart.getCatName(), newPart.getDbName(),
newPart.getTableName());
- MPartition oldp = getMPartition(catName, dbname, name, part_vals, table);
- return alterPartitionNoTxn(catName, dbname, name, oldp, newPart,
- validWriteIds, oldCd, table);
- }
-
private Partition alterPartitionNoTxn(String catName, String dbname,
String name, MPartition oldp, Partition newPart,
String validWriteIds,
@@ -1873,15 +1867,15 @@ private Partition alterPartitionNoTxn(String catName,
String dbname,
catName = normalizeIdentifier(catName);
name = normalizeIdentifier(name);
dbname = normalizeIdentifier(dbname);
+ if (oldp == null) {
+ throw new InvalidObjectException("partition does not exist.");
+ }
MPartition newp = convertToMPart(newPart, table);
MColumnDescriptor oldCD = null;
MStorageDescriptor oldSD = oldp.getSd();
if (oldSD != null) {
oldCD = oldSD.getCD();
}
- if (newp == null) {
- throw new InvalidObjectException("partition does not exist.");
- }
oldp.setValues(newp.getValues());
oldp.setPartitionName(newp.getPartitionName());
boolean isTxn = TxnUtils.isTransactionalTable(table.getParameters());
@@ -1969,7 +1963,7 @@ protected List<Partition> alterPartitionsInternal(MTable
table,
throw new MetaException("Invalid DB name : " + tmpPart.getDbName());
}
if (!tmpPart.getTableName().equalsIgnoreCase(tblName)) {
- throw new MetaException("Invalid table name : " + tmpPart.getDbName());
+ throw new MetaException("Invalid table name : " +
tmpPart.getTableName());
}
}
return new GetListHelper<TableName, Partition>(this, null) {
@@ -2004,8 +1998,9 @@ private List<Partition> alterPartitionsViaJdo(MTable
table, List<String> partNam
mPartitionList = (List<MPartition>) query.executeWithArray(tblName,
dbName, partNames, catName);
pm.retrieveAll(mPartitionList);
- if (mPartitionList.size() > newParts.size()) {
- throw new MetaException("Expecting only one partition but more than
one partitions are found.");
+ if (mPartitionList.size() != newParts.size()) {
+ throw new MetaException("Expected " + newParts.size() + " partitions
but found "
+ + mPartitionList.size());
}
Map<List<String>, MPartition> mPartsMap = new HashMap();
@@ -2017,8 +2012,9 @@ private List<Partition> alterPartitionsViaJdo(MTable
table, List<String> partNam
AtomicReference<MColumnDescriptor> oldCdRef = new AtomicReference<>();
for (Partition tmpPart : newParts) {
oldCdRef.set(null);
+ MPartition mPart = mPartsMap.get(tmpPart.getValues());
Partition result = alterPartitionNoTxn(catName, dbName, tblName,
- mPartsMap.get(tmpPart.getValues()), tmpPart, queryWriteIdList,
oldCdRef, table);
+ mPart, tmpPart, queryWriteIdList, oldCdRef, table);
results.add(result);
if (oldCdRef.get() != null) {
oldCds.add(oldCdRef.get());
@@ -2051,7 +2047,8 @@ public List<Partition> getPartitionsByFilter(TableName
tableName, GetPartitionsA
@Override
protected boolean canUseDirectSql() throws MetaException {
- return getDirectSql().generateSqlFilterForPushdown(catName, dbName,
tblName, partitionKeys, tree, null, filter);
+ return getDirectSql().generateSqlFilterForPushdown(catName, dbName,
tblName, partitionKeys,
+ tree, null, filter);
}
@Override
@@ -2106,8 +2103,8 @@ protected boolean canUseDirectSql() throws MetaException {
// if the filter mode is BY_EXPR initialize the filter and generate
the expression tree
// if there are more than one filter string we AND them together
initExpressionTree();
- return
getDirectSql().generateSqlFilterForPushdown(table.getCatName(),
table.getDbName(), table.getTableName(),
- table.getPartitionKeys(), tree, null, filter);
+ return
getDirectSql().generateSqlFilterForPushdown(table.getCatName(),
table.getDbName(),
+ table.getTableName(), table.getPartitionKeys(), tree, null,
filter);
}
// BY_VALUES and BY_NAMES are always supported
return true;
@@ -2363,7 +2360,8 @@ private List<MPartition>
listMPartitionsWithProjection(List<String> fieldNames,
}
@Override
- public int getNumPartitionsByFilter(TableName tableName, String filter)
throws MetaException, NoSuchObjectException {
+ public int getNumPartitionsByFilter(TableName tableName, String filter)
+ throws MetaException, NoSuchObjectException {
final ExpressionTree exprTree =
org.apache.commons.lang3.StringUtils.isNotEmpty(filter)
? PartFilterExprUtil.parseFilterTree(filter) :
ExpressionTree.EMPTY_TREE;
@@ -2383,7 +2381,8 @@ protected String describeResult() {
@Override
protected boolean canUseDirectSql() throws MetaException {
- return getDirectSql().generateSqlFilterForPushdown(catName, dbName,
tblName, partitionKeys, exprTree, null, filter);
+ return getDirectSql().generateSqlFilterForPushdown(catName, dbName,
tblName, partitionKeys,
+ exprTree, null, filter);
}
@Override
@@ -2397,7 +2396,8 @@ protected Integer getJdoResult() throws MetaException,
NoSuchObjectException {
}.run(false);
}
- private Integer getNumPartitionsViaOrmFilter(String catName, String dbName,
String tblName, ExpressionTree tree, boolean isValidatedFilter,
List<FieldSchema> partitionKeys)
+ private Integer getNumPartitionsViaOrmFilter(String catName, String dbName,
String tblName, ExpressionTree tree,
+ boolean isValidatedFilter, List<FieldSchema> partitionKeys)
throws MetaException {
Map<String, Object> params = new HashMap<>();
String jdoFilter = makeQueryFilterString(catName, dbName, tblName, tree,
@@ -2852,7 +2852,8 @@ public int getObjectCount(String fieldName, String
objName) {
@Override
public long updateParameterWithExpectedValue(Table table, String key, String
expectedValue, String newValue)
throws MetaException, NoSuchObjectException {
- return new GetHelper<TableName, Long>(this, new
TableName(table.getCatName(), table.getDbName(), table.getTableName())) {
+ return new GetHelper<TableName, Long>(this,
+ new TableName(table.getCatName(), table.getDbName(),
table.getTableName())) {
@Override
protected String describeResult() {
return "Affected rows";
@@ -3109,8 +3110,8 @@ private static StorageDescriptor
convertToStorageDescriptor(
msd.getLocation(), msd.getInputFormat(), msd.getOutputFormat(), msd
.isCompressed(), msd.getNumBuckets(),
(!isAcidTable) ? convertToSerDeInfo(msd.getSerDeInfo(), conf, true)
- : new SerDeInfo(msd.getSerDeInfo().getName(),
msd.getSerDeInfo().getSerializationLib(), Collections.emptyMap()),
- bucList , orderList, sdParams);
+ : new SerDeInfo(msd.getSerDeInfo().getName(),
msd.getSerDeInfo().getSerializationLib(),
+ Collections.emptyMap()), bucList , orderList, sdParams);
if (!isAcidTable) {
skewedInfo = new SkewedInfo(convertList(msd.getSkewedColNames()),
convertToSkewedValues(msd.getSkewedColValues()),
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/ExecutionContextTestUtils.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/ExecutionContextTestUtils.java
new file mode 100644
index 00000000000..1423788e64b
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/ExecutionContextTestUtils.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.metastore;
+
+import org.apache.hadoop.hive.metastore.metastore.PersistenceManagerProxy;
+import org.datanucleus.ExecutionContext;
+import org.datanucleus.api.jdo.JDOPersistenceManager;
+import org.datanucleus.cache.Level1Cache;
+import org.datanucleus.state.DNStateManager;
+
+import javax.jdo.PersistenceManager;
+
+/**
+ * Helpers for inspecting DataNucleus L1 (persistence context) cache in unit
tests.
+ */
+public final class ExecutionContextTestUtils {
+
+ private ExecutionContextTestUtils() {
+ }
+
+ public static ExecutionContext getExecutionContext(PersistenceManager pm) {
+ if (pm instanceof JDOPersistenceManager) {
+ return ((JDOPersistenceManager) pm).getExecutionContext();
+ }
+ if (pm instanceof PersistenceManagerProxy.ExecutionContextReference) {
+ return ((PersistenceManagerProxy.ExecutionContextReference)
pm).getExecutionContext();
+ }
+ throw new IllegalArgumentException("Unsupported PersistenceManager: " +
pm.getClass());
+ }
+
+ public static int countCachedInstances(PersistenceManager pm, Class<?>
clazz) {
+ Level1Cache l1Cache = getExecutionContext(pm).getLevel1Cache();
+ int count = 0;
+ for (DNStateManager stateManager : l1Cache.values()) {
+ if (clazz.isInstance(stateManager.getObject())) {
+ count++;
+ }
+ }
+ return count;
+ }
+}
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestPersistenceContextEviction.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestPersistenceContextEviction.java
new file mode 100644
index 00000000000..d798e238c75
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/metastore/TestPersistenceContextEviction.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.metastore;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.ExecutionContextTestUtils;
+import org.apache.hadoop.hive.metastore.HMSHandler;
+import org.apache.hadoop.hive.metastore.MetaStoreTestUtils;
+import org.apache.hadoop.hive.metastore.ObjectStore;
+import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.NotificationEvent;
+import org.apache.hadoop.hive.metastore.api.NotificationEventRequest;
+import org.apache.hadoop.hive.metastore.api.NotificationEventResponse;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
+import org.apache.hadoop.hive.metastore.messaging.EventMessage;
+import org.apache.hadoop.hive.metastore.model.MNotificationLog;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import javax.jdo.PersistenceManager;
+import javax.jdo.Query;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Verifies batched metastore operations evict loaded JDO entities from the
persistence context.
+ * Without eviction, long-lived RawStore instances (for example the DB
notification cleaner thread)
+ * accumulate deleted entities in the L1 cache and can OOM.
+ */
+@Category(MetastoreUnitTest.class)
+public class TestPersistenceContextEviction {
+ private static final int BATCH_SIZE = 3;
+ private static final int NUM_EVENTS = 12;
+
+ private ObjectStore objectStore;
+ private Configuration conf;
+ private PersistenceManager pm;
+
+ @Before
+ public void setUp() throws Exception {
+ conf = MetastoreConf.newMetastoreConf();
+ MetastoreConf.setBoolVar(conf, ConfVars.HIVE_IN_TEST, true);
+ MetastoreConf.setLongVar(conf,
MetastoreConf.ConfVars.EVENT_CLEAN_MAX_EVENTS, BATCH_SIZE);
+ MetastoreConf.setLongVar(conf, ConfVars.RAWSTORE_PARTITION_BATCH_SIZE,
BATCH_SIZE);
+ MetaStoreTestUtils.setConfForStandloneMode(conf);
+
+ String currentUrl = MetastoreConf.getVar(conf, ConfVars.CONNECT_URL_KEY);
+ currentUrl = currentUrl.replace(MetaStoreServerUtils.JUNIT_DATABASE_PREFIX,
+ String.format("%s_%s", MetaStoreServerUtils.JUNIT_DATABASE_PREFIX,
UUID.randomUUID()));
+ MetastoreConf.setVar(conf, ConfVars.CONNECT_URL_KEY, currentUrl);
+
+ objectStore = new ObjectStore();
+ objectStore.setConf(conf);
+ HMSHandler.createDefaultCatalog(objectStore, new Warehouse(conf));
+ pm = objectStore.createRawStoreBundle().getPersistentManager();
+ }
+
+ @Test
+ public void testExecutionContextCountsLoadedNotificationEvents() throws
MetaException {
+ insertNotificationEvents(5, "payload");
+
+ objectStore.openTransaction();
+ try {
+ Query query = pm.newQuery(MNotificationLog.class);
+ List<MNotificationLog> events = (List<MNotificationLog>) query.execute();
+ pm.retrieveAll(events);
+ Assert.assertTrue("expected loaded events to remain in the persistence
context",
+ ExecutionContextTestUtils.countCachedInstances(pm,
MNotificationLog.class) >= 5);
+ } finally {
+ objectStore.rollbackTransaction();
+ }
+
+ Assert.assertEquals(0, ExecutionContextTestUtils.countCachedInstances(pm,
MNotificationLog.class));
+ }
+
+ @Test
+ public void testCleanNotificationEventsEvictsCachedEntities() throws
MetaException {
+ insertNotificationEvents(NUM_EVENTS, "x".repeat(50));
+
+ objectStore.openTransaction();
+ try {
+ objectStore.cleanNotificationEvents(0);
+ Assert.assertEquals("batched notification cleanup retains deleted events
in the L1 cache", NUM_EVENTS,
+ ExecutionContextTestUtils.countCachedInstances(pm,
MNotificationLog.class));
+ } finally {
+ objectStore.commitTransaction();
+ }
+
+ Assert.assertEquals("batched notification cleanup must not retain deleted
events in the L1 cache", 0,
+ ExecutionContextTestUtils.countCachedInstances(pm,
MNotificationLog.class));
+ NotificationEventResponse response = objectStore.getNextNotification(new
NotificationEventRequest());
+ Assert.assertEquals(0, response.getEventsSize());
+ }
+
+ private void insertNotificationEvents(int count, String message) throws
MetaException {
+ for (int i = 0; i < count; i++) {
+ NotificationEvent event = new NotificationEvent(0, 0,
+ EventMessage.EventType.CREATE_DATABASE.toString(), message);
+ objectStore.addNotificationEvent(event);
+ }
+ }
+}
