### Question:
TopNOperator implements AdapterWorkProcessorOperator { @Override public void addInput(Page page) { addPage(page); } private  TopNOperator(
            MemoryTrackingContext memoryTrackingContext,
            Optional<WorkProcessor<Page>> sourcePages,
            List<Type> types,
            int n,
            List<Integer> sortChannels,
            List<SortOrder> sortOrders,
            TypeOperators typeOperators); static OperatorFactory createOperatorFactory(
            int operatorId,
            PlanNodeId planNodeId,
            List<? extends Type> types,
            int n,
            List<Integer> sortChannels,
            List<SortOrder> sortOrders,
            TypeOperators typeOperators); @Override WorkProcessor<Page> getOutputPages(); @Override boolean needsInput(); @Override void addInput(Page page); @Override void finish();  }### Answer:
@Test public void testExceedMemoryLimit() throws Exception { List<Page> input = rowPagesBuilder(BIGINT) .row(1L) .build(); DriverContext smallDiverContext = createTaskContext(executor, scheduledExecutor, TEST_SESSION, DataSize.ofBytes(1)) .addPipelineContext(0, true, true, false) .addDriverContext(); OperatorFactory operatorFactory = topNOperatorFactory( ImmutableList.of(BIGINT), 100, ImmutableList.of(0), ImmutableList.of(ASC_NULLS_LAST)); try (Operator operator = operatorFactory.createOperator(smallDiverContext)) { operator.addInput(input.get(0)); operator.getOutput(); fail("must fail because of exceeding local memory limit"); } catch (ExceededMemoryLimitException ignore) { } }