NIFI-10366: Make Default Run Duration configurable (#6310)

NIFI-10366: Make Default Run Duration configurable
This commit is contained in:
Lehel Boér 2022-08-30 20:38:56 +02:00 committed by GitHub
parent bae659cb9e
commit 28fc030ef5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 60 additions and 1 deletions

View File

@ -0,0 +1,40 @@
/*
* 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.nifi.annotation.behavior;
import java.time.Duration;
public enum DefaultRunDuration {
NO_BATCHING(Duration.ZERO),
TWENTY_FIVE_MILLIS(Duration.ofMillis(25)),
FIFTY_MILLIS(Duration.ofMillis(50)),
ONE_HUNDRED_MILLIS(Duration.ofMillis(100)),
TWO_HUNDRED_FIFTY_MILLIS(Duration.ofMillis(250)),
FIVE_HUNDRED_MILLIS(Duration.ofMillis(500)),
ONE_SECOND(Duration.ofSeconds(1)),
TWO_SECONDS(Duration.ofSeconds(2));
private final Duration duration;
DefaultRunDuration(final Duration duration) {
this.duration = duration;
}
public Duration getDuration() {
return duration;
}
}

View File

@ -41,11 +41,13 @@ import java.lang.annotation.Target;
* ProcessSession.commit() to ensure data is persisted before deleting the data
* from a remote source.
*
* When the defaultDuration parameter is set, the processor is created with the supplied duration time, which can be adjusted afterwards.
* The supplied values can be selected from {@link org.apache.nifi.annotation.behavior.DefaultRunDuration}.
*/
@Documented
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface SupportsBatching {
DefaultRunDuration defaultDuration() default DefaultRunDuration.NO_BATCHING;
}

View File

@ -16,9 +16,11 @@
*/
package org.apache.nifi.controller;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.ClassUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.behavior.RequiresInstanceClassLoading;
import org.apache.nifi.annotation.behavior.SupportsBatching;
import org.apache.nifi.annotation.configuration.DefaultSettings;
import org.apache.nifi.bundle.Bundle;
import org.apache.nifi.bundle.BundleCoordinate;
@ -320,6 +322,8 @@ public class ExtensionBuilder {
}
applyDefaultSettings(procNode);
applyDefaultRunDuration(procNode);
return procNode;
}
@ -359,6 +363,19 @@ public class ExtensionBuilder {
}
}
private void applyDefaultRunDuration(final ProcessorNode processorNode) {
try {
final Class<?> procClass = processorNode.getProcessor().getClass();
final SupportsBatching sb = procClass.getAnnotation(SupportsBatching.class);
if (sb != null) {
processorNode.setRunDuration(sb.defaultDuration().getDuration().toMillis(), TimeUnit.MILLISECONDS);
}
} catch (final Exception ex) {
logger.error("Set Default Run Duration failed", ex);
}
}
private ControllerServiceNode createControllerServiceNode() throws ClassNotFoundException, IllegalAccessException, InstantiationException, InitializationException {
final ClassLoader ctxClassLoader = Thread.currentThread().getContextClassLoader();
try {