From 278e5967fdc6c72910431de9e4145e54b2101f85 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Wed, 19 Aug 2026 14:22:18 +0200 Subject: [PATCH] HIVE-29820: Refactor ObjectCache in RecordProcessors --- .../hive/ql/exec/ObjectCacheFactory.java | 16 +- .../hive/ql/exec/tez/MapRecordProcessor.java | 32 +--- .../ql/exec/tez/MergeFileRecordProcessor.java | 20 +- .../hive/ql/exec/tez/RecordProcessor.java | 90 +++++++-- .../ql/exec/tez/ReduceRecordProcessor.java | 31 +-- .../hive/ql/exec/tez/TestRecordProcessor.java | 180 ++++++++++++++++++ 6 files changed, 271 insertions(+), 98 deletions(-) create mode 100644 ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestRecordProcessor.java diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/ObjectCacheFactory.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/ObjectCacheFactory.java index c23a116eeae7..132a7de057f4 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/ObjectCacheFactory.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/ObjectCacheFactory.java @@ -19,6 +19,7 @@ package org.apache.hadoop.hive.ql.exec; +import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; import org.slf4j.Logger; @@ -26,7 +27,6 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.llap.io.api.LlapProxy; -import org.apache.hadoop.hive.llap.io.api.LlapProxy; import org.apache.hadoop.hive.ql.exec.tez.LlapObjectCache; /** @@ -92,15 +92,11 @@ private static boolean isLlapCacheEnabled(Configuration conf, boolean isPlanCach private static ObjectCache getLlapObjectCache(String queryId) { // If order of events (i.e. dagstart and fragmentstart) was guaranteed, we could just // create the cache when dag starts, and blindly return it to execution here. - if (queryId == null) throw new RuntimeException("Query ID cannot be null"); - ObjectCache result = llapQueryCaches.get(queryId); - if (result != null) return result; - result = new LlapObjectCache(); - ObjectCache old = llapQueryCaches.putIfAbsent(queryId, result); - if (old == null) { - LOG.info("Created object cache for " + queryId); - } - return (old != null) ? old : result; + Objects.requireNonNull(queryId, "Query ID cannot be null"); + return llapQueryCaches.computeIfAbsent(queryId, k -> { + LOG.info("Created object cache for {}", k); + return new LlapObjectCache(); + }); } public static void removeLlapQueryCache(String queryId) { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java index 67ef4cb5ac01..aad1f345266c 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java @@ -46,8 +46,6 @@ import org.apache.hadoop.hive.ql.exec.HashTableDummyOperator; import org.apache.hadoop.hive.ql.exec.MapOperator; import org.apache.hadoop.hive.ql.exec.MapredContext; -import org.apache.hadoop.hive.ql.exec.ObjectCache; -import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.OperatorUtils; import org.apache.hadoop.hive.ql.exec.TezDummyStoreOperator; @@ -93,20 +91,14 @@ public class MapRecordProcessor extends RecordProcessor { private final ExecMapperContext execContext; private MapWork mapWork; private List mergeWorkList; - private final List cacheKeys = new ArrayList<>(); - private final List dynamicValueCacheKeys = new ArrayList<>(); - private final ObjectCache cache, dynamicValueCache; // is this part of the query-based compaction process private final boolean isInCompaction; public MapRecordProcessor(final JobConf jconf, final ProcessorContext context) throws Exception { super(jconf, context); - String queryId = HiveConf.getVar(jconf, HiveConf.ConfVars.HIVE_QUERY_ID); if (LlapProxy.isDaemon()) { setLlapOfFragmentId(context); } - cache = ObjectCacheFactory.getCache(jconf, queryId, true); - dynamicValueCache = ObjectCacheFactory.getCache(jconf, queryId, false, true); execContext = new ExecMapperContext(jconf); execContext.setJc(jconf); isInCompaction = CompactorUtil.COMPACTOR.equalsIgnoreCase( @@ -129,12 +121,10 @@ void init(MRTaskReporter mrReporter, String key = processorContext.getTaskVertexName() + MAP_PLAN_KEY; - cacheKeys.add(key); - // create map and fetch operators if (!isInCompaction) { - mapWork = cache.retrieve(key, () -> Utilities.getMapWork(jconf)); + mapWork = planCache.retrieve(key, () -> Utilities.getMapWork(jconf)); } else { // During query-based compaction, we don't want to retrieve old MapWork from the cache, we want a new mapper // and new UDF validate_acid_sort_order instance for each bucket, otherwise validate_acid_sort_order will fail. @@ -160,11 +150,10 @@ void init(MRTaskReporter mrReporter, } key = processorContext.getTaskVertexName() + prefix; - cacheKeys.add(key); checkAbortCondition(); mergeWorkList.add( - (MapWork) cache.retrieve(key, () -> Utilities.getMergeWork(jconf, prefix))); + (MapWork) planCache.retrieve(key, () -> Utilities.getMergeWork(jconf, prefix))); } } @@ -307,9 +296,8 @@ void init(MRTaskReporter mrReporter, checkAbortCondition(); String valueRegistryKey = DynamicValue.DYNAMIC_VALUE_REGISTRY_CACHE_KEY; // On LLAP dynamic value registry might already be cached. - final DynamicValueRegistryTez registryTez = dynamicValueCache.retrieve(valueRegistryKey, - () -> new DynamicValueRegistryTez()); - dynamicValueCacheKeys.add(valueRegistryKey); + final DynamicValueRegistryTez registryTez = + dynamicValueCache.retrieve(valueRegistryKey, () -> new DynamicValueRegistryTez()); RegistryConfTez registryConf = new RegistryConfTez(jconf, mapWork, processorContext, inputs); registryTez.init(registryConf); @@ -442,17 +430,7 @@ void close(){ setAborted(execContext.getIoCxt().getIOExceptions()); } - if (cache != null) { - for (String k: cacheKeys) { - cache.release(k); - } - } - - if (dynamicValueCache != null) { - for (String k: dynamicValueCacheKeys) { - dynamicValueCache.release(k); - } - } + releaseCache(); // detecting failed executions by exceptions thrown by the operator tree try { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java index b380d907d814..d5df4272cd64 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java @@ -25,9 +25,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.exec.MapredContext; -import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.OperatorUtils; import org.apache.hadoop.hive.ql.exec.Utilities; @@ -59,11 +57,9 @@ public class MergeFileRecordProcessor extends RecordProcessor { protected Operator mergeOp; private ExecMapperContext execContext = null; protected static final String MAP_PLAN_KEY = "__MAP_PLAN__"; - private String cacheKey; private MergeFileWork mfWork; MRInputLegacy mrInput = null; private final Object[] row = new Object[2]; - org.apache.hadoop.hive.ql.exec.ObjectCache cache; public MergeFileRecordProcessor(final JobConf jconf, final ProcessorContext context) { super(jconf, context); @@ -94,20 +90,10 @@ void init( .initialize(); } - String queryId = HiveConf.getVar(jconf, HiveConf.ConfVars.HIVE_QUERY_ID); - cache = ObjectCacheFactory.getCache(jconf, queryId, true); - try { execContext.setJc(jconf); - cacheKey = MAP_PLAN_KEY; - - MapWork mapWork = (MapWork) cache.retrieve(cacheKey, new Callable() { - @Override - public Object call() { - return Utilities.getMapWork(jconf); - } - }); + MapWork mapWork = (MapWork) planCache.retrieve(MAP_PLAN_KEY, (Callable) () -> Utilities.getMapWork(jconf)); Utilities.setMapWork(jconf, mapWork); if (mapWork instanceof MergeFileWork) { @@ -162,9 +148,7 @@ void run() throws Exception { @Override void close() { - if (cache != null && cacheKey != null) { - cache.release(cacheKey); - } + releaseCache(); // check if there are IOExceptions if (!isAborted()) { diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java index 64bec2c1b70c..b287cf5c3289 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java @@ -22,8 +22,11 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.concurrent.Callable; +import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.exec.ObjectCache; +import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; import org.apache.hadoop.hive.ql.exec.Utilities; import org.apache.hadoop.hive.ql.exec.tez.TezProcessor.TezKVOutputCollector; import org.apache.hadoop.hive.ql.log.PerfLogger; @@ -59,9 +62,32 @@ public abstract class RecordProcessor extends InterruptibleProcessing { protected PerfLogger perfLogger = SessionState.getPerfLogger(); protected String CLASS_NAME = RecordProcessor.class.getName(); + protected final String queryId; + + /** + * Per-processor plan cache — no daemon-wide sharing. Sharing would race on + * per-fragment operator state that {@code initializeOp()} resets (HIVE-14433: + * {@code FileSinkOperator.fsp}, {@code VectorGroupByOperator.aggregator}, + * {@code Operator.childOperatorsArray}, {@code VectorTopNKeyOperator} filter + * state, ...) and yields NPE / {@code FileAlreadyExistsException}. + */ + protected final TrackedCache planCache; + + /** + * Dynamic value cache. On LLAP this is the per-query daemon-wide + * {@link LlapObjectCache}, so dynamic values computed once (e.g. broadcast + * hash tables, DPP registries) are reused across fragments of the same query. + */ + protected final TrackedCache dynamicValueCache; + public RecordProcessor(JobConf jConf, ProcessorContext processorContext) { this.jconf = jConf; this.processorContext = processorContext; + this.queryId = HiveConf.getVar(jConf, HiveConf.ConfVars.HIVE_QUERY_ID); + this.planCache = new TrackedCache( + ObjectCacheFactory.getCache(jConf, queryId, true, false)); + this.dynamicValueCache = new TrackedCache( + ObjectCacheFactory.getCache(jConf, queryId, false, true)); } /** @@ -98,26 +124,60 @@ protected void createOutputMap() { } } - public List getMergeWorkList(final JobConf jconf, String key, String queryId, - ObjectCache cache, List cacheKeys) throws HiveException { + /** + * Release every key retrieved through the plan and dynamic-value caches. + * A no-op for {@link LlapObjectCache} (which relies on soft references), but + * preserved for correctness against other {@link ObjectCache} implementations. + */ + protected void releaseCache() { + planCache.releaseAll(); + dynamicValueCache.releaseAll(); + } + + public List getMergeWorkList(final JobConf jconf) throws HiveException { String prefixes = jconf.get(DagUtils.TEZ_MERGE_WORK_FILE_PREFIXES); - if (prefixes != null) { - List mergeWorkList = new ArrayList<>(); + if (prefixes == null) { + return null; + } + List mergeWorkList = new ArrayList<>(); + for (final String prefix : prefixes.split(",")) { + if (prefix.isEmpty()) { + continue; + } + mergeWorkList.add(planCache.retrieve(prefix, () -> Utilities.getMergeWork(jconf, prefix))); + } + return mergeWorkList; + } + + /** + * An {@link ObjectCache} paired with the set of keys retrieved through it, so + * {@link #releaseAll()} releases exactly those keys at close time. All retrievals + * that need to be released should go through {@link #retrieve} — calling + * {@link ObjectCache#retrieve} directly on the underlying cache bypasses the + * tracking and leaks the key. + */ + protected static final class TrackedCache { + private final ObjectCache cache; + private final List keys = new ArrayList<>(); - for (final String prefix : prefixes.split(",")) { - if (prefix.isEmpty()) { - continue; - } + TrackedCache(ObjectCache cache) { + this.cache = cache; + } - key = prefix; - cacheKeys.add(key); + /** Retrieve (or compute) the value for {@code key}, tracking it for release. */ + T retrieve(String key, Callable fn) throws HiveException { + keys.add(key); + return cache.retrieve(key, fn); + } - mergeWorkList.add(cache.retrieve(key, () -> Utilities.getMergeWork(jconf, prefix))); + /** Release every key retrieved through this wrapper. Null-safe on the underlying cache. */ + void releaseAll() { + if (cache == null) { + return; + } + for (String k : keys) { + cache.release(k); } - - return mergeWorkList; - } else { - return null; } } } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java index c37f0434e6c3..3b622ed9656c 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java @@ -30,12 +30,9 @@ import org.apache.hadoop.hive.llap.LlapUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.exec.DummyStoreOperator; import org.apache.hadoop.hive.ql.exec.HashTableDummyOperator; import org.apache.hadoop.hive.ql.exec.MapredContext; -import org.apache.hadoop.hive.ql.exec.ObjectCache; -import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.OperatorUtils; import org.apache.hadoop.hive.ql.exec.Utilities; @@ -56,8 +53,6 @@ import org.apache.tez.runtime.api.ProcessorContext; import org.apache.tez.runtime.api.Reader; -import com.google.common.collect.Lists; - /** * Process input from tez LogicalInput and write output - for a map plan * Just pump the records through the query plan. @@ -67,13 +62,9 @@ public class ReduceRecordProcessor extends RecordProcessor { private static final String REDUCE_PLAN_KEY = "__REDUCE_PLAN__"; - private final ObjectCache cache, dynamicValueCache; - private ReduceWork reduceWork; private final List mergeWorkList; - private final List cacheKeys; - private final List dynamicValueCacheKeys = new ArrayList<>(); private final Map connectOps = new TreeMap<>(); private final Map tagToReducerMap = new HashMap<>(); @@ -87,16 +78,11 @@ public class ReduceRecordProcessor extends RecordProcessor { public ReduceRecordProcessor(final JobConf jconf, final ProcessorContext context) throws Exception { super(jconf, context); - String queryId = HiveConf.getVar(jconf, HiveConf.ConfVars.HIVE_QUERY_ID); - cache = ObjectCacheFactory.getCache(jconf, queryId, true); - dynamicValueCache = ObjectCacheFactory.getCache(jconf, queryId, false, true); - String cacheKey = processorContext.getTaskVertexName() + REDUCE_PLAN_KEY; - cacheKeys = Lists.newArrayList(cacheKey); - reduceWork = cache.retrieve(cacheKey, () -> Utilities.getReduceWork(jconf)); + reduceWork = planCache.retrieve(cacheKey, () -> Utilities.getReduceWork(jconf)); Utilities.setReduceWork(jconf, reduceWork); - mergeWorkList = getMergeWorkList(jconf, cacheKey, queryId, cache, cacheKeys); + mergeWorkList = getMergeWorkList(jconf); } @Override @@ -161,7 +147,6 @@ void init(MRTaskReporter mrReporter, Map inputs, Map new DynamicValueRegistryTez()); - dynamicValueCacheKeys.add(valueRegistryKey); RegistryConfTez registryConf = new RegistryConfTez(jconf, reduceWork, processorContext, inputs); registryTez.init(registryConf); checkAbortCondition(); @@ -339,17 +324,7 @@ private List getShuffleInputs(Map inputs) th @Override void close() { - if (cache != null) { - for (String key : cacheKeys) { - cache.release(key); - } - } - - if (dynamicValueCache != null) { - for (String k : dynamicValueCacheKeys) { - dynamicValueCache.release(k); - } - } + releaseCache(); try { boolean abort = isAborted(); diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestRecordProcessor.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestRecordProcessor.java new file mode 100644 index 000000000000..fe895d3e30f3 --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestRecordProcessor.java @@ -0,0 +1,180 @@ +/* + * 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.ql.exec.tez; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.io.api.LlapProxy; +import org.apache.hadoop.hive.ql.exec.ObjectCache; +import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory; +import org.apache.hadoop.hive.ql.exec.tez.RecordProcessor.TrackedCache; +import org.apache.hadoop.mapred.JobConf; +import org.apache.tez.mapreduce.processor.MRTaskReporter; +import org.apache.tez.runtime.api.LogicalInput; +import org.apache.tez.runtime.api.LogicalOutput; +import org.apache.tez.runtime.api.ProcessorContext; +import org.junit.Test; + +/** + * Tests for {@link RecordProcessor}. Behaviour exercised here lives on the + * base class and is inherited by every subclass — {@link MapRecordProcessor}, + * {@link ReduceRecordProcessor}, and {@link MergeFileRecordProcessor} — so + * covering it once here is enough. + */ +public class TestRecordProcessor { + + private static Callable constant(T value) { + return () -> value; + } + + /** + * {@code TrackedCache.retrieve} must delegate to the underlying cache and + * return the value the underlying cache produced — no wrapping, no + * substitution. Every subclass assigns this return value directly to a + * plan field ({@code mapWork}, {@code reduceWork}), so any drift here would + * be a silent bug. + */ + @Test + public void retrieveReturnsUnderlyingValue() throws Exception { + ObjectCache backend = mock(ObjectCache.class); + Object sentinel = new Object(); + when(backend.retrieve(anyString(), any())).thenReturn(sentinel); + TrackedCache planCache = new TrackedCache(backend); + + Object got = planCache.retrieve("k", constant(sentinel)); + + assertEquals(sentinel, got); + verify(backend).retrieve(any(), any()); + } + + /** + * The plan cache and the dynamic-value cache must track their keys + * independently — releasing one must not touch the other. Map/Reduce + * processors rely on this invariant when they retrieve their plan plus + * the dynamic value registry from two separate {@link TrackedCache}s. + */ + @Test + public void planAndDynamicValueCachesAreIndependent() throws Exception { + ObjectCache planBackend = mock(ObjectCache.class); + ObjectCache dvBackend = mock(ObjectCache.class); + when(planBackend.retrieve(anyString(), any())).thenReturn("plan"); + when(dvBackend.retrieve(anyString(), any())).thenReturn("dv"); + + TrackedCache planCache = new TrackedCache(planBackend); + TrackedCache dynamicValueCache = new TrackedCache(dvBackend); + + planCache.retrieve("Map 1__MAP_PLAN__", constant("plan")); + dynamicValueCache.retrieve("dyn-values", constant("dv")); + + planCache.releaseAll(); + verify(planBackend).release("Map 1__MAP_PLAN__"); + verify(dvBackend, never()).release(anyString()); // dv registry key not touched yet + // dv still holds its key until we call its own releaseAll + dynamicValueCache.releaseAll(); + verify(dvBackend).release("dyn-values"); + } + + /** + * Regression guard for the plan cache's LLAP wiring. Two processors of the + * same query on the same LLAP daemon must NOT share a deserialised plan — + * sharing would let concurrent fragments race on the per-fragment operator + * state ({@link org.apache.hadoop.hive.ql.exec.FileSinkOperator#fsp}, + * {@code VectorGroupByOperator.aggregator}, {@code VectorTopNKeyOperator} + * filter state, {@link org.apache.hadoop.hive.ql.exec.Operator#childOperatorsArray}) + * that {@code initializeOp()} resets, producing the + * {@code FileAlreadyExistsException} / {@code NullPointerException} class of + * failures HIVE-14433 documents. + * + *

The test drives the real code path — it flips {@link LlapProxy} into + * daemon mode, constructs two {@link StubRecordProcessor} instances through + * the production constructor, and calls {@code retrieve()} on their + * {@code planCache} fields. Each must load its own copy and the loader must + * fire once per processor. If {@code RecordProcessor}'s constructor ever + * flips its plan cache back to {@code llapCacheAlwaysEnabled=true}, this + * test fails. + */ + @Test + public void plansAreNotSharedAcrossProcessorsOnLlap() throws Exception { + String queryId = "record-processor-test-" + System.nanoTime(); + LlapProxy.setDaemon(true); + try { + JobConf conf = new JobConf(); + HiveConf.setVar(conf, HiveConf.ConfVars.HIVE_EXECUTION_ENGINE, "tez"); + HiveConf.setVar(conf, HiveConf.ConfVars.HIVE_QUERY_ID, queryId); + ProcessorContext ctx = mock(ProcessorContext.class); + + RecordProcessor procOne = new StubRecordProcessor(conf, ctx); + RecordProcessor procTwo = new StubRecordProcessor(conf, ctx); + + String key = "Reducer 2__REDUCE_PLAN__"; + AtomicInteger loaderInvocations = new AtomicInteger(); + Callable loader = () -> { + loaderInvocations.incrementAndGet(); + return new Object(); + }; + + Object planFromOne = procOne.planCache.retrieve(key, loader); + Object planFromTwo = procTwo.planCache.retrieve(key, loader); + + assertEquals("each processor must load its own plan", + 2, loaderInvocations.get()); + assertNotSame("plans must not be shared across processors on LLAP", + planFromOne, planFromTwo); + } finally { + ObjectCacheFactory.removeLlapQueryCache(queryId); + LlapProxy.setDaemon(false); + } + } + + /** + * Minimal {@link RecordProcessor} subclass so the test can construct one + * without pulling in a live Map/Reduce operator tree. The parent constructor + * is the only behaviour under test. + */ + private static final class StubRecordProcessor extends RecordProcessor { + StubRecordProcessor(JobConf jconf, ProcessorContext context) { + super(jconf, context); + } + + @Override + void init(MRTaskReporter mrReporter, + Map inputs, Map outputs) { + } + + @Override + void run() { + } + + @Override + void close() { + } + } +}