### Question:
KeepAliveDataSource implements DirectDataSource { @Override public <T> ModelInput<T> openInput( DataDefinition<T> definition, DirectInputFragment fragment, Counter counter) throws IOException, InterruptedException { ModelInput<T> input = entity.openInput(definition, fragment, counter); return new WrappedModelInput<>(input, heartbeat, counter); } KeepAliveDataSource(DirectDataSource entity, long interval); @Override String path(String basePath, ResourcePattern resourcePattern); @Override String path(String basePath); @Override List<DirectInputFragment> findInputFragments(
            DataDefinition<T> definition,
            String basePath,
            ResourcePattern resourcePattern); @Override ModelInput<T> openInput(
            DataDefinition<T> definition,
            DirectInputFragment fragment,
            Counter counter); @Override ModelOutput<T> openOutput(
            OutputAttemptContext context,
            DataDefinition<T> definition,
            String basePath,
            String resourcePath,
            Counter counter); @Override List<ResourceInfo> list(
            String basePath,
            ResourcePattern resourcePattern,
            Counter counter); @Override boolean delete(
            String basePath,
            ResourcePattern resourcePattern,
            boolean recursive,
            Counter counter); @Override void setupAttemptOutput(OutputAttemptContext context); @Override void commitAttemptOutput(OutputAttemptContext context); @Override void cleanupAttemptOutput(OutputAttemptContext context); @Override void setupTransactionOutput(OutputTransactionContext context); @Override void commitTransactionOutput(OutputTransactionContext context); @Override void cleanupTransactionOutput(OutputTransactionContext context); @Override Optional<T> findProperty(Class<T> propertyType);  }### Answer:
@Test public void testOpenInput() throws Exception { try (ModelInput<Object> input = ds.openInput(null, null, counter)) { assertKeepAlive(true); } assertKeepAlive(false); }