Lynxus 开发者指南

在这种情况下，如果“Use File”属性设置为 `true`，则除非“File to Use”属性设置为有效的文件名，否则处理器无效。如果“Use File”设置为 `true` 且“File to Use”未设置值，则处理器无效（因为“File to Use”属性是必需的）。如果“Use File”设置为 `true` 且“File to Use”已设置值但指定的文件不存在，处理器也将无效，因为根据验证器“File to Use”属性无效。

然而，如果“Use File”属性设置为 `false`，则“File to Use”属性被认为具有未满足的依赖关系。因此，“File to Use”属性将不会在验证中被考虑。所以，如果“Use File”设置为 `false` 且“File to Use”没有值，处理器仍然有效（即使“File to Use”是必需的，也仅在“Use File”为 `true` 时才需要）。同样，如果“File to Use”设置为不存在的文件名（如 `/file/that/does/not/exist`），只要“Use File”属性设置为 `false`，处理器仍然有效。

此外，除非“Use File”设置为 `true`，否则“File to Use”属性甚至不会在 Lynxus UI 中显示。

但是，有时单独验证处理器的属性是不够的。为此，AbstractProcessor 公开了 `customValidate` 方法。该方法接受一个 `ValidationContext` 类型的参数。该方法的返回值是一个 `ValidationResult` 对象的 `Collection`，描述在验证过程中发现的任何问题。只应返回那些 `isValid` 方法返回 `false` 的 ValidationResult 对象。仅当所有属性根据其关联的验证器和允许值都有效时，才会调用此方法。也就是说，仅当所有属性本身都有效时才会调用此方法，此方法允许对处理器的整体配置进行验证。


==== 响应配置更改

有时希望处理器在其属性更改时立即做出反应。`onPropertyModified` 方法允许处理器做到这一点。当用户更改处理器的属性值时，将为每个修改的属性调用 `onPropertyModified` 方法。该方法接受三个参数：指示哪个属性被修改的 PropertyDescriptor、旧值和新值。如果属性之前没有值，则第二个参数为 `null`。如果属性被移除，则第三个参数为 `null`。重要的是要注意，无论值是否有效，都会调用此方法。仅当值实际被修改时才调用此方法，而不是在用户更新处理器但未更改其值时调用。在调用此方法时，保证调用此方法的线程是当前在处理器中执行代码的唯一线程，除非处理器本身创建了自己的线程。


==== 执行工作

当处理器有工作要做时，框架通过调用其 `onTrigger` 方法来调度它。该方法接受两个参数：`ProcessContext` 和 `ProcessSession`。`onTrigger` 方法的第一步通常是通过调用 ProcessSession 上的 `get` 方法之一来获取要执行工作的 FlowFile。对于从外部源将数据摄入 Lynxus 的处理器，此步骤被跳过。然后处理器可以自由检查 FlowFile 属性；添加、删除或修改属性；读取或修改 FlowFile 内容；以及将 FlowFile 转移到适当的 Relationship。


==== 处理器的触发时机

处理器的 `onTrigger` 方法仅在其被调度运行且处理器存在工作可做时才会被调用。如果满足以下任一条件，则认为处理器存在工作：

- 目标为该处理器的连接在其队列中至少有一个 FlowFile
- 处理器没有传入连接
- 处理器标注了 @TriggerWhenEmpty 注解

有多个因素会影响处理器的 `onTrigger` 方法何时被调用。首先，除非用户已配置处理器运行，否则不会触发处理器。如果处理器被调度运行，框架会定期（周期由用户在用户界面中配置）检查是否有工作要做，如上所述。如果是，框架将检查处理器的下游目标。如果处理器的任何传出连接已满，默认情况下，处理器将不会被调度运行。

但是，`@TriggerWhenAnyDestinationAvailable` 注解可以添加到处理器的类中。在这种情况下，要求更改为只需要一个下游目标"可用"（如果连接的队列未满，则认为目标"可用"），而不是要求所有下游目标都可用。

与处理器调度相关的还有 `@TriggerSerially` 注解。使用此注解的处理器永远不会有一个以上的线程同时运行 `onTrigger` 方法。但需要注意的是，执行代码的线程可能在不同调用之间发生变化。因此，仍必须注意确保处理器是线程安全的！



=== 组件生命周期

Lynxus API 通过 Java 注解提供生命周期支持。`cn.org.acal.lynxus.annotation.lifecycle` 包包含多个用于生命周期管理的注解。以下注解可以应用于 Lynxus 组件中的 Java 方法，以指示框架何时应调用这些方法。在讨论组件生命周期时，我们将 Lynxus 组件定义为处理器、控制器服务、参数提供程序或报告任务。

==== @OnAdded

`@OnAdded` 注解使方法在组件创建后立即被调用。组件的 `initialize` 方法（如果继承 `AbstractProcessor` 则为 `init` 方法）将在组件构造后被调用，然后是带有 `@OnAdded` 注解的方法。如果任何带有 `@OnAdded` 注解的方法抛出异常，将向用户返回错误，并且该组件不会被添加到数据流中。此外，带有此注解的其他方法也不会被调用。此方法在组件的生命周期内仅调用一次。带有此注解的方法必须不接受任何参数。

==== @OnEnabled

`@OnEnabled` 注解用于指示每当控制器服务被启用时应调用的方法。任何带有此注解的方法都将在用户每次启用服务时被调用。此外，每次 Lynxus 重新启动时，如果 Lynxus 配置为“自动恢复状态”且服务已启用，则将调用该方法。

如果带有此注解的方法抛出 Throwable，将为该组件发出日志消息和公告。在这种情况下，服务将保持“启用中”状态且不可使用。延迟后将再次调用带有此注解的所有方法。在所有带有此注解的方法都正常返回而未抛出任何异常之前，服务不会被投入使用。

使用此注解的方法必须接受 0 个参数或一个 `cn.org.acal.lynxus.controller.ConfigurationContext` 类型的参数。

请注意，如果将此注解应用于报告任务、参数提供程序或处理器，它将被忽略。对于控制器服务，启用和禁用被视为生命周期事件，因为该操作使其可被或不可被其他组件使用。然而，对于处理器和报告任务，这些不是生命周期事件，而是一种允许在启动或停止一组组件时排除某个组件的机制。

==== @OnRemoved

`@OnRemoved` 注解使方法在组件从数据流中移除之前被调用。这允许在移除组件之前清理资源。带有此注解的方法必须不接受任何参数。如果带有此注解的方法抛出异常，组件仍将被移除。

==== @OnScheduled

此注解指示每当组件被调度运行时应调用该方法。由于控制器服务不被调度，在控制器服务上使用此注解没有意义且不会被执行。它应仅用于处理器和报告任务。如果带有此注解的任何方法抛出异常，带有此注解的其他方法将不会被调用，并且将向用户显示通知。在这种情况下，带有`@OnUnscheduled` 注解的方法将被触发，然后是带有 `@OnStopped` 注解的方法（在此状态下，如果这些方法中的任何一个抛出异常，这些异常将被忽略）。然后组件将在一段时间内让出执行，称为“管理让出持续时间”，这是在 `lynxus.properties` 文件中配置的值。最后，该过程将重新开始，直到所有带有 `@OnScheduled` 注解的方法都正常返回而未抛出任何异常。带有此注解的方法可以接受零个参数或一个参数。如果使用单参数变体，当组件是处理器时，参数必须是 `ProcessContext` 类型；当组件是报告任务时，参数必须是 `ConfigurationContext` 类型。

==== @OnUnscheduled

带有此注解的方法将在处理器或报告任务不再被调度运行时被调用。此时，可能仍有许多线程在处理器的 `onTrigger` 方法中处于活动状态。如果这样的方法抛出异常，将生成日志消息，异常将被忽略，带有此注解的其他方法仍将被调用。带有此注解的方法可以接受零个参数或一个参数。如果使用单参数变体，当组件是处理器时，参数必须是 `ProcessContext` 类型；当组件是报告任务时，参数必须是 `ConfigurationContext` 类型。


==== @OnStopped

带有此注解的方法将在处理器或报告任务不再被调度运行且所有线程都已从 `onTrigger` 方法返回时被调用。如果这样的方法抛出异常，将生成日志消息，异常将被忽略；带有此注解的其他方法仍将被调用。带有此注解的方法允许接受 0 或 1 个参数。如果使用参数，当组件是报告任务时，它必须是 ConfigurationContext 类型；当组件是处理器时，它必须是 ProcessContext 类型。


==== @OnShutdown

任何带有 `@OnShutdown` 注解的方法将在 Lynxus 成功关闭时被调用。如果这样的方法抛出异常，将生成日志消息，异常将被忽略，带有此注解的其他方法仍将被调用。带有此注解的方法必须不接受任何参数。注意：虽然 Lynxus 会尝试在所有使用此注解的组件上调用带有此注解的方法，但这并不总是可行的。例如，进程可能会被意外终止，在这种情况下它没有机会调用这些方法。因此，虽然使用此注解的方法可以用于清理资源，但不应依赖它们来处理关键业务逻辑。

=== 组件通知

Lynxus API 通过 Java 注解提供通知支持。`cn.org.acal.lynxus.annotation.notification` 包包含多个用于通知管理的注解。以下注解可以应用于 Lynxus 组件中的 Java 方法，以指示框架何时应调用这些方法。在讨论组件通知时，我们将 Lynxus 组件定义为*处理器*、*控制器服务*或*报告任务*。

==== @OnPrimaryNodeStateChange

`@OnPrimaryNodeStateChange` 注解使方法在集群中主节点的状态发生变化后立即被调用。带有此注解的方法应不接受参数或接受一个 `PrimaryNodeState` 类型的参数。`PrimaryNodeState` 提供有关更改内容的上下文，以便组件可以采取适当的操作。`PrimaryNodeState` 枚举有两个可能的值：`ELECTED_PRIMARY_NODE`（接收此状态的节点已被选为 Lynxus 集群的主节点），或 `PRIMARY_NODE_REVOKED`（接收此状态的节点曾是主节点，但其主节点角色已被撤销）。

[[restricted]]
=== 受限

`Restricted` 类注解已被移除（参见 https://issues.apache.org/jira/browse/NIP-24[NIP-24]）。Lynxus 不再基于此注解将组件视为单独的“受限”类别进行授权。请在组件的能力描述中记录敏感能力，并根据部署需要使用标准授权策略。

[[state_manager]]
=== 状态管理器

通过 ProcessContext、ReportingContext 和 ControllerServiceInitializationContext，组件能够调用 `getStateManager()` 方法。此状态管理器负责提供一个简单的 API 用于存储和检索状态。此机制旨在为开发者提供一种非常简便的方式来存储一组键/值对、检索这些值以及原子性地更新它们。状态可以存储在本地节点或集群中的所有节点上。但重要的是要注意，此机制仅旨在提供一种存储非常“简单”状态的机制。因此，该 API 仅允许存储和检索 `Map<String, String>` 以及原子性地替换整个 Map。此外，目前唯一支持的用于存储集群范围状态的实现由 ZooKeeper 支持。因此，整个状态 Map 在序列化后的大小必须小于 1 MB。尝试存储超过此限制的数据将导致抛出异常。如果处理器管理状态所需的交互比这更复杂（例如，必须存储和检索大量数据，或者必须单独存储和获取各个键），则应使用不同的机制（例如，与外部数据库通信）。


[[state_scope]]
==== 作用域
与状态管理器通信时，所有方法调用都需要提供 Scope。此 Scope 为 `Scope.LOCAL` 或 `Scope.CLUSTER`。如果 Lynxus 在集群中运行，此 Scope 向框架提供有关操作应如何进行的重要信息。

如果使用 `Scope.CLUSTER` 存储状态，则集群中的所有节点将与相同的状态存储机制通信。如果使用 `Scope.LOCAL` 存储和检索状态，则每个节点将看到不同的状态表示。

还值得注意的是，如果 Lynxus 配置为作为独立实例运行，而不是在集群中运行，则始终使用 `Scope.LOCAL` 的 scope。这样做是为了允许 Lynxus 组件的开发者以一种一致的方式编写代码，而不必担心 Lynxus 实例是否是集群的。开发者应该假设实例是集群的并相应地编写代码。


==== 存储和检索状态

状态通过 StateManager 的 `getState`、`setState`、`replace` 和 `clear` 方法存储。所有这些方法都需要提供 Scope。需要注意的是，使用 Local scope 存储的状态与使用 Cluster scope 存储的状态完全不同。如果处理器使用 `Scope.CLUSTER` scope 存储键为 _My Key_ 的值，然后尝试使用 `Scope.LOCAL` scope 检索该值，则检索到的值将为 `null`（除非也使用 `Scope.CLUSTER` scope 存储了相同键的值）。每个处理器的状态与其他处理器的状态隔离存储。

因此，两个处理器不能共享相同的状态。然而，在某些情况下，在两个不同类型的处理器或两个相同类型的处理器之间共享状态是非常必要的。这可以通过使用控制器服务来实现。通过在控制器服务中存储和检索状态，多个处理器可以使用相同的控制器服务，并且状态可以通过控制器服务的 API 公开。


==== 单元测试
Lynxus 的 Mock 框架提供了大量的工具来执行处理器的单元测试。处理器单元测试通常从 `TestRunner` 类开始。因此，`TestRunner` 类包含自己的 `getStateManager` 方法。然而，返回的 StateManager 是一个特定类型：`MockStateManager`。此实现除了 `StateManager` 接口定义的方法外，还提供了多个方法，帮助开发者更轻松地开发单元测试。

首先，`MockStateManager` 实现了 `StateManager` 接口，因此可以在单元测试中检查所有状态。此外，`MockStateManager` 公开了一组 `assert*` 方法来断言状态是否按预期设置。`MockStateManager` 还提供了在特定 `Scope` 的状态被更新时立即使单元测试失败的能力。




=== 报告处理器活动

处理器负责报告其活动，使用户能够了解其数据发生了什么。处理器应通过 ComponentLog 记录事件，可以通过 InitializationContext 或调用 `AbstractProcessor` 的 `getLogger` 方法访问。

此外，处理器应使用通过 ProcessSession 的 `getProvenanceReporter` 方法获取的 `ProvenanceReporter` 接口。ProvenanceReporter 应用于指示每次从外部源接收内容或发送到外部位置的情况。ProvenanceReporter 还有用于报告 FlowFile 被克隆、分叉或修改以及多个 FlowFile 合并为单个 FlowFile 的方法，以及将 FlowFile 与某个其他标识符关联。然而，这些功能的报告不太关键，因为框架能够检测这些情况并代表处理器发出适当的事件。但是，处理器开发者的最佳实践是发出这些事件，因为在代码中明确发出了这些事件，并且开发者能够为事件提供额外的详细信息，例如操作所花费的时间或关于所执行操作的相关信息。如果处理器发出了事件，框架将不会发出重复的事件。相反，它始终假设处理器开发者比框架更了解处理器上下文中发生的事情。但是，框架可能会发出不同的事件。例如，如果处理器同时修改了 FlowFile 的内容和属性，然后仅发出 ATTRIBUTES_MODIFIED 事件，框架将发出 CONTENT_MODIFIED 事件。如果为该 FlowFile 发出了任何其他事件（无论是处理器还是框架发出的），框架将不会发出 ATTRIBUTES_MODIFIED 事件。这是因为所有 <<provenance_events>> 都知道事件发生前 FlowFile 的属性以及处理该 FlowFile 后产生的属性，因此 ATTRIBUTES_MODIFIED 通常被认为是多余的，并且会导致 FlowFile 血统的呈现非常冗长。然而，如果从处理器的角度来看该事件被认为是相关的，处理器可以将其与其他事件一起发出。







== 组件文档编写

Lynxus 尝试通过用户界面从 Lynxus 应用程序本身向用户提供大量文档，使用户体验尽可能简单和方便。为此，处理器开发者必须向框架提供这些文档。Lynxus 公开了几种不同的机制来向框架提供文档。


=== 文档化属性

可以通过调用 PropertyDescriptor 构建器的 `description` 方法来文档化各个属性，如下所示：

[source,java]
----
public static final PropertyDescriptor MY_PROPERTY = new PropertyDescriptor.Builder()
  .name("My Property")
  .description("Description of the Property")
  ...
  .build();
----

如果属性要提供一组允许值，这些值将在 UI 的下拉字段中呈现给用户。每个值还可以附带描述：

[source,java]
----
public static final AllowableValue EXTENSIVE = new AllowableValue("Extensive", "Extensive",
  "Everything will be logged - use with caution!");
public static final AllowableValue VERBOSE = new AllowableValue("Verbose", "Verbose",
  "Quite a bit of logging will occur");
public static final AllowableValue REGULAR = new AllowableValue("Regular", "Regular",
  "Typical logging will occur");

public static final PropertyDescriptor LOG_LEVEL = new PropertyDescriptor.Builder()
  .name("Amount to Log")
  .description("How much the Processor should log")
  .allowableValues(REGULAR, VERBOSE, EXTENSIVE)
  .defaultValue(REGULAR.getValue())
  ...
  .build();
----


=== 文档化关系

处理器 Relationship 的文档化方式与属性非常相似——通过调用 Relationship 构建器的 `description` 方法：

[source,java]
----
public static final Relationship MY_RELATIONSHIP = new Relationship.Builder()
  .name("My Relationship")
  .description("This relationship is used only if the Processor fails to process the data.")
  .build();
----


=== 文档化能力和关键词

`cn.org.acal.lynxus.annotation.documentation` 包提供了可用于文档化组件的 Java 注解。CapabilityDescription 注解可以添加到处理器、报告任务或控制器服务中，旨在提供组件功能的简要描述。Tags 注解有一个 `value` 变量，定义为字符串数组。因此，它的使用方式是提供多个值作为逗号分隔的字符串列表，用花括号括起来。然后通过允许用户基于标签（即关键字）过滤组件来将这些值整合到 UI 中。此外，UI 提供了标签云，允许用户选择要过滤的标签。标签云中最大的标签是在该 Lynxus 实例的组件中存在最多的标签。以下是使用这些注解的示例：

[source, java]
----
@Tags({"example", "documentation", "developer guide", "processor", "tags"})
@CapabilityDescription("Example Processor that provides no real functionality but is provided" +
  " for an example in the Developer Guide")
public static final ExampleProcessor extends Processor {
    ...
}
----

=== 文档化 FlowFile 属性交互

很多时候，处理器会期望在传入的 FlowFile 上设置某些 FlowFile 属性，以便处理器正常运行。在其他情况下，处理器可能会在传出的 FlowFile 上更新或创建 FlowFile 属性。处理器开发者可以使用 `ReadsAttribute` 和 `WritesAttribute` 文档注解来文档化这两种行为。这些属性用于生成文档，使用户更好地了解处理器如何与数据流交互。

注意：由于 Java 7 不支持类型上的重复注解，您可能需要使用 `ReadsAttributes` 和 `WritesAttributes` 来指示处理器读取或写入多个 FlowFile 属性。此注解只能应用于处理器。下面列出了一个示例：

[source, java]
----
@WritesAttributes({ @WritesAttribute(attribute = "invokehttp.status.code", description = "The status code that is returned"),
        @WritesAttribute(attribute = "invokehttp.status.message", description = "The status message that is returned"),
        @WritesAttribute(attribute = "invokehttp.response.body", description = "The response body"),
        @WritesAttribute(attribute = "invokehttp.request.url", description = "The request URL"),
        @WritesAttribute(attribute = "invokehttp.tx.id", description = "The transaction ID that is returned after reading the response"),
        @WritesAttribute(attribute = "invokehttp.remote.dn", description = "The DN of the remote server") })
public final class InvokeHTTP extends AbstractProcessor {
----

=== 文档化相关组件
处理器和控制器服务经常相互关联。有时是 put/get 关系，如 `PutFile` 和 `GetFile`。有时处理器使用控制器服务，如 `InvokeHTTP` 和 `StandardSSLContextService`。有时一个控制器服务使用另一个，如 `MapCacheClientService` 和 `MapCacheServer`。这些扩展点的开发者可以使用 `SeeAlso` 标签关联这些不同的组件。此注解在文档中链接这些组件。`SeeAlso` 可以应用于处理器、控制器服务、参数提供程序和报告任务。下面列出了如何执行此操作的示例：

[source, java]
----
@SeeAlso(GetFile.class)
public class PutFile extends AbstractProcessor {
----

=== 高级文档

当上述文档方法不够时，Lynxus 提供了通过“用法”文档向用户公开更高级文档的能力。当用户右键单击处理器时，Lynxus 在上下文菜单中提供“用法”菜单项。此外，UI 在右上角公开了“帮助”链接，从中可以找到相同的用法信息。

处理器的高级文档以名为 `additionalDetails.md` 的 Markdown 文件形式提供。此文件应存在于一个目录中，该目录的名称是处理器的完全限定名，此目录的父目录应命名为 `docs` 并存在于处理器的 jar 的根目录中。此文件将从生成的 HTML 文件链接，该 HTML 文件将包含所有能力、关键字、属性描述和 Relationship 信息，因此无需重复这些内容。这是一个提供关于此处理器的功能、期望和产生的数据类型以及期望和产生的 FlowFile 属性的详细说明的地方。由于此文档采用 HTML 格式，您可以包含图片和表格来最好地描述此组件。相同的方法可用于为处理器、控制器服务、参数提供程序和报告任务提供高级文档。

[[provenance_events]]
== 溯源事件

溯源报告的不同事件类型有：

[options="header"]
|======================
|溯源事件        |描述
|ADDINFO                 |表示添加额外信息的溯源事件，例如与新的 URI 或 UUID 的新链接
|ATTRIBUTES_MODIFIED     |表示 FlowFile 的属性以某种方式被修改。当同时报告另一个事件时不需要此事件，因为另一个事件已包含所有 FlowFile 属性
|CLONE                   |表示 FlowFile 是其父 FlowFile 的精确副本
|CONTENT_MODIFIED        |表示 FlowFile 的内容以某种方式被修改。使用此事件类型时，建议提供有关内容如何修改的详细信息
|CREATE                  |表示 FlowFile 是从未从远程系统或外部进程接收的数据生成的
|DOWNLOAD                |表示 FlowFile 的内容被用户或外部实体下载
|DROP                    |表示对象因对象过期以外的原因结束其生命周期的溯源事件
|EXPIRE                  |表示对象因未及时处理的溯源事件
|FETCH                   |表示 FlowFile 的内容被某个外部资源的内容覆盖。这类似于 RECEIVE 事件，但不同之处在于 RECEIVE 事件旨在用作将 FlowFile 引入系统的事件，而 FETCH 用于表示现有 FlowFile 的内容被覆盖
|FORK                    |表示一个或多个 FlowFile 是从父 FlowFile 派生的
|JOIN                    |表示单个 FlowFile 是从多个父 FlowFile 合并而来的
|RECEIVE                 |表示从外部进程接收数据的溯源事件。此事件类型预期是 FlowFile 的第一个事件。因此，从外部源接收数据并使用该数据替换现有 FlowFile 内容的处理器应使用 FETCH 事件类型，而不是 RECEIVE 事件类型
|REMOTE_INVOCATION       |表示请求对外部端点进行远程调用（例如删除远程资源）。外部端点可能存在于远程或本地系统中，但对 Lynxus 来说是外部的
|REPLAY                  |表示重放 FlowFile 的溯源事件。事件的 UUID 表示正在重放的原始 FlowFile 的 UUID。该事件包含一个父 UUID（也是正在重放的 FlowFile 的 UUID）和一个子 UUID（将重新排队进行处理的新创建的 FlowFile 的 UUID）
|ROUTE                   |表示 FlowFile 被路由到指定的 Relationship，并提供有关 FlowFile 为何被路由到此 Relationship 的信息
|SEND                    |表示向外部进程发送数据的溯源事件
|UPLOAD                  |表示将外部数据上传到外部系统的溯源事件
|UNKNOWN                 |表示溯源事件的类型未知，因为尝试访问该事件的用户无权知道该类型
|======================


== 常见处理器模式

虽然 Lynxus 用户可以使用许多不同的处理器，但其中绝大多数属于几种常见的设计模式之一。下面我们讨论这些模式、何时适用这些模式、我们遵循这些模式的原因，以及应用这些模式时需要注意的事项。请注意，下面讨论的模式和建议是一般性指南，而非硬性规则。


[[ingress]]
=== 数据入口

将数据摄入 Lynxus 的处理器有一个名为 `success` 的 Relationship。此处理器通过 ProcessSession 的 `create` 方法生成新的 FlowFile，并且不从传入连接中的 FlowFile。处理器名称以“Get”或“Listen”开头，取决于它是轮询外部源还是公开某个外部源可以连接的接口。名称以用于通信的协议结尾。遵循此模式的处理器包括 `GetFile`、`GetSFTP`、`ListenHTTP` 和 `GetHTTP`。

此处理器可以在使用 `@OnScheduled` 注解的方法中创建或初始化连接池。但是，由于通信问题可能阻止连接建立或导致连接终止，连接本身不在此时创建。相反，连接在 `onTrigger` 方法中从池中创建或租用。

此处理器的 `onTrigger` 方法首先尝试从连接池租用连接，如果不可能则创建到外部服务的连接。当外部源没有可用数据时，处理器调用 ProcessContext 的 `yield` 方法并返回，以避免此处理器持续运行并无益地消耗资源。否则，此处理器然后通过 ProcessSession 的 `create` 方法创建一个 FlowFile，并为其分配适当的文件名和路径（通过添加 `filename` 和 `path` 属性），以及任何其他可能适当的属性。通过 ProcessSession 的 `write` 方法获取 FlowFile 内容的 OutputStream，传入一个新的 OutputStreamCallback（通常是匿名内部类）。在此回调中，处理器能够写入 FlowFile 并将外部资源的内容流式传输到 FlowFile 的 OutputStream。如果希望将 InputStream 的全部内容写入 FlowFile，使用 ProcessSession 的 `importFrom` 方法可能比 `write` 方法更方便。

当此处理器预期接收许多小文件时，建议在提交会话之前从单个会话创建多个 FlowFile。通常，这允许框架更高效地处理新创建的 FlowFile 的内容。

此处理器生成一个溯源事件，指示它已接收数据并指定数据来源。此处理器应记录 FlowFile 的创建，以便在必要时可以通过分析日志确定 FlowFile 的来源。

此处理器确认数据的接收和/或从外部源移除数据，以防止接收重复文件。*这仅在 FlowFile 创建时的 ProcessSession 已提交后才执行！* 不遵循此原则可能导致数据丢失，因为在会话提交之前重启 Lynxus 将导致临时文件被删除。但请注意，使用此方法可能会接收重复数据，因为应用程序可能在提交会话后、确认或从外部源移除数据之前重启。然而，通常情况下，潜在的数据重复优于潜在的数据丢失。连接最终被返回或添加到连接池，取决于连接是从连接池租用的还是在 `onTrigger` 方法中创建的。

如果出现通信问题，连接通常会被终止并且不会返回（或添加）到连接池。与远程系统的连接会被断开，连接池会在标注了 `@OnStopped` 注解的方法中关闭，以便回收资源。


=== 数据出口

将数据发布到外部源的处理器有两个 Relationship：`success` 和 `failure`。处理器名称以“Put”开头，后跟用于数据传输的协议。处理器遵循此模式的包括 `PutEmail`、`PutSFTP` 和 `PostHTTP`（注意名称不以“Put”开头，因为这会导致混淆，因为 PUT 和 POST 在处理 HTTP 时有特殊含义）。

此处理器可以在使用 `@OnScheduled` 注解的方法中创建或初始化连接池。但是，由于通信问题可能阻止连接建立或导致连接终止，连接本身不在此时创建。相反，连接在 `onTrigger` 方法中从池中创建或租用。

`onTrigger` 方法首先通过 `get` 方法从 ProcessSession 获取 FlowFile。如果没有可用的 FlowFile，方法在不获取到远程资源连接的情况下返回。

如果至少有一个 FlowFile 可用，处理器从连接池获取连接（如果可能），否则创建新连接。如果处理器既无法从连接池租用连接也无法创建新连接，则 FlowFile 被路由到 `failure`，事件被记录，方法返回。

如果获取了连接，处理器通过调用 ProcessSession 的 `read` 方法并传入 InputStreamCallback（通常是匿名内部类）获取 FlowFile 内容的 InputStream，并在该回调中将 FlowFile 的内容传输到目标位置。事件与传输文件所花的时间和文件传输的数据速率一起被记录。通过 ProcessSession 的 `getProvenanceReporter` 方法获取报告器并调用报告器的 `send` 方法，向 ProvenanceReporter 报告 SEND 事件。连接被返回或添加到连接池，取决于连接是从池租用的还是由 `onTrigger` 方法新创建的。

如果出现通信问题，连接通常会被终止并且不会返回（或添加）到连接池。如果将数据发送到远程资源时出现问题，处理错误的期望方法取决于一些考虑因素。如果问题与网络条件有关，FlowFile 通常被路由到 `failure`。FlowFile 不会被惩罚，因为数据不一定有问题。与<<ingress>>处理器的情况不同，我们通常不在 ProcessContext 上调用 `yield`。这是因为在摄入的情况下，FlowFile 在处理器能够执行其功能。然而，对于 Put 处理器，数据流管理员可以选择将 `failure` 路由到不同的处理器。这允许在一个系统出现问题时使用“备份”系统，或者用于在多个系统之间进行负载分配。

如果出现数据相关的问题，应采取两种方法之一。首先，如果问题可能会自行解决，则 FlowFile 被惩罚然后路由到 `failure`。例如，PutFTP 就是这种情况，当 FlowFile 因文件命名冲突而无法传输时。假设文件最终会从目录中移除，以便新文件可以传输。因此，我们惩罚 FlowFile 并路由到 `failure`，以便稍后重试。另一种情况，如果数据本身有问题（例如数据不符合某些要求的规范），则可以采取不同的方法。在这种情况下，将 `failure` 关系拆分为 `failure` 和 `communications failure` 关系可能是有利的。这允许数据流管理员分别确定如何处理每种情况。在这些情况下，重要的是在创建 Relationship 时通过在“描述”中阐明来充分记录两个 Relationship 之间的差异。

与远程系统的连接会被断开，连接池在标注了 `@OnStopped` 的方法中关闭，以便回收资源。


=== 基于内容的路由（一对多）

基于内容路由数据的处理器有两种形式之一：将传入的 FlowFile 路由到恰好一个目标，或将传入数据路由到 0 个或多个目标。这里，我们讨论第一种情况。

此处理器有两个 Relationship：`matched` 和 `unmatched`。如果期望特定的数据格式，处理器还将有一个 `failure` Relationship，用于输入不符合预期格式的情况。处理器公开一个指示路由标准的属性。

如果指定路由标准的属性需要处理（例如编译正则表达式），则此处理尽可能在标注了 `@OnScheduled` 的方法中完成。然后将结果存储在标记为 `volatile` 的成员变量中。

`onTrigger` 方法获取单个 FlowFile。该方法通过 ProcessSession 的 `read` 方法读取 FlowFile 的内容，在数据流式传输时评估匹配标准。然后处理器根据标准是否匹配确定 FlowFile 应路由到 `matched` 还是 `unmatched`，并将 FlowFile 路由到适当的 Relationship。

然后处理器发出一个 Provenance ROUTE 事件，指示处理器将 FlowFile 路由到了哪个 Relationship。

此处理器标注了 `cn.org.acal.lynxus.annotation.behavior` 包中的 `@SideEffectFree` 和 `@SupportsBatching` 注解。


=== 基于内容的路由（一对多）

如果处理器要将单个 FlowFile 路由到多个 Relationship，此处理器将与上述基于内容路由数据的处理器略有不同。此处理器通常具有由用户动态定义的 Relationship 以及一个 `unmatched` Relationship。

为了使用户能够定义额外的属性，必须重写 `getSupportedDynamicPropertyDescriptor` 方法。此方法返回一个具有提供名称的 PropertyDescriptor 和一个适用的验证器，以确保用户指定的匹配标准有效。

在此处理器中，`getRelationships` 方法返回的 Relationship 集合是一个标记为 `volatile` 的成员变量。此集合最初使用一个名为 `unmatched` 的 Relationship 构建。`onPropertyModified` 方法被重写，以便在添加或移除属性时，创建一个同名的新 Relationship。如果处理器具有非用户定义的属性，检查指定属性是否为用户定义是很重要的。这可以通过调用传递给此方法的 PropertyDescriptor 的 `isDynamic` 方法来实现。如果此属性是动态的，则创建一个新的 Relationship 集合，并将之前的 Relationship 集合复制到其中。这个新集合要么添加新创建的 Relationship，要么从中移除，取决于是否向处理器添加了新属性或移除了属性（通过检查此函数的第三个参数是否为 `null` 来检测属性移除）。然后更新持有 Relationship 集合的成员变量以指向这个新集合。

如果指定路由标准的属性需要处理（例如编译正则表达式），则该处理尽可能在标注了 `@OnScheduled` 的方法中完成。然后将结果存储在标记为 `volatile` 的成员变量中。此成员变量通常是 `Map` 类型，其中键为 `Relationship` 类型，值的类型由属性值的处理结果定义。

`onTrigger` 方法通过 ProcessSession 的 `get` 方法获取 FlowFile。如果没有可用的 FlowFile，则立即返回。否则，创建一个 Relationship 类型的 Set。该方法通过 ProcessSession 的 `read` 方法读取 FlowFile 的内容，在数据流式传输时评估每个匹配标准。对于任何匹配的标准，与该匹配标准关联的 Relationship 被添加到 Relationship 集合中。

读取 FlowFile 的内容后，该方法检查 Relationship 集合是否为空。如果是，则向原始 FlowFile 添加一个属性以指示其路由到的 Relationship，并将其路由到 `unmatched`。这会被记录，发出一个 Provenance ROUTE 事件，然后方法返回。如果集合的大小等于 1，则向原始 FlowFile 添加一个属性以指示其路由到的 Relationship，并将其路由到集合中指定的 Relationship。这会被记录，为该 FlowFile 发出一个 Provenance ROUTE 事件，然后方法返回。

如果集合包含多个 Relationship，处理器会为每个 Relationship 创建 FlowFile 的克隆，第一个除外。这通过 ProcessSession 的 `clone` 方法完成。无需报告 CLONE 溯源事件，因为框架会自动处理。原始 FlowFile 和每个克隆都被路由到其适当的 Relationship，并带有指示 Relationship 名称的属性。为每个 FlowFile 发出一个 Provenance ROUTE 事件。这会被记录，然后方法返回。

此处理器标注了来自 `cn.org.acal.lynxus.annotation.behavior` 包的 `@SideEffectFree` 和 `@SupportsBatching` 注解。


=== 基于内容的流路由（一对多）

前面关于基于内容路由（一对多）的描述提供了一个抽象，用于创建一个非常强大的处理器。然而，它假设每个 FlowFile 将作为一个整体路由到零个或多个 Relationship。如果传入的数据格式是许多不同信息的“流”——而我们希望将此流的不同部分发送到不同的 Relationship？例如，假设我们想要一个 RouteCSV 处理器，配置了多个正则表达式。如果 CSV 文件中的一行匹配某个正则表达式，该行应包含在发往关联 Relationship 的出站 FlowFile 中。如果一个正则表达式关联到 Relationship“has-apples”并且该正则表达式匹配 FlowFile 中的 1,000 行，则“has-apples”Relationship 应有一个包含 1,000 行的出站 FlowFile。如果另一个正则表达式关联到 Relationship“has-oranges”并且该正则表达式匹配 FlowFile 中的 50 行，则“has-oranges”Relationship 应有一个包含 50 行的出站 FlowFile。即，一个 FlowFile 进来，两个 FlowFile 出去。两个 FlowFile 可能包含原始 FlowFile 中的一些相同文本行，也可能完全不同。这就是本节要讨论的处理器类型。

此处理器名称以“Route”开头，以它路由的数据类型名称结尾。在我们的示例中，我们路由 CSV 数据，所以处理器命名为 RouteCSV。此处理器支持动态属性。每个用户定义的属性的名称映射到 Relationship 的名称。属性的值是“匹配标准”所需的格式。在我们的示例中，属性的值必须是有效的正则表达式。

此处理器维护一个内部 `ConcurrentMap`，其中键是 `Relationship`，值的类型取决于匹配标准的格式。在我们的示例中，我们维护一个 `ConcurrentMap<Relationship, Pattern>`。此处理器重写 `onPropertyModified` 方法。如果提供给此方法的新值（第三个参数）为 null，则从 ConcurrentMap 中移除属性名（第一个参数）定义的 Relationship。否则，处理新值（在我们的示例中，通过调用 `Pattern.compile(newValue)`）并将此值添加到 ConcurrentMap，键再次是由属性名指定的 Relationship。

此处理器将重写 `customValidate` 方法。在此方法中，它将从 `ValidationContext` 检索所有属性并计算动态 PropertyDescriptor 的数量（通过调用 PropertyDescriptor 的 `isDynamic()`）。如果动态 PropertyDescriptor 的数量为 0，这表明用户没有添加任何 Relationship，因此处理器返回一个 `ValidationResult`，指示处理器无效，因为它没有添加任何 Relationship。

处理器在其 `getRelationships` 方法被调用时返回用户指定的所有 Relationship，并还将返回一个 `unmatched` Relationship。由于此处理器必须读写内容仓库（这可能相对昂贵），如果此处理器预计用于非常高的数据量，添加一个允许用户指定是否关心不匹配任何匹配标准的数据的属性可能是有利的。

当 `onTrigger` 方法被调用时，处理器通过 `ProcessSession.get` 获取 FlowFile。如果没有可用数据，处理器返回。否则，处理器创建一个 `Map<Relationship, FlowFile>`。我们将此 Map 称为 `flowFileMap`。处理器通过调用 `ProcessSession.read` 读取传入的 FlowFile，并提供一个 `InputStreamCallback`。在回调中，处理器从 FlowFile 读取第一条数据。然后处理器根据每条匹配标准评估此数据。如果特定标准（在我们的示例中是正则表达式）匹配，处理器从 `flowFileMap` 获取属于相应 Relationship 的 FlowFile。如果此 Relationship 在 Map 中还没有 FlowFile，处理器通过调用 `session.create(incomingFlowFile)` 创建新的 FlowFile，然后将新 FlowFile 添加到 `flowFileMap`。然后处理器通过调用带有 `OutputStreamCallback` 的 `session.append` 将此数据写入 FlowFile。在此 OutputStreamCallback 中，我们可以访问新 FlowFile 的 OutputStream，因此能够将数据写入新 FlowFile。然后从 OutputStreamCallback 返回。遍历所有匹配标准后，如果都不匹配，我们对 `unmatched` Relationship 执行与上述相同的操作（除非用户配置为不写出未匹配的数据）。由于我们已经调用了 `session.append`，我们有了 FlowFile 的新版本。因此，我们需要更新 `flowFileMap`，将 Relationship 与新 FlowFile 关联。

如果在任何时候抛出异常，我们需要将传入的 FlowFile 路由到 `failure`。我们还需要移除每个新创建的 FlowFile，因为我们不会将它们传输到任何地方。我们可以通过调用 `session.remove(flowFileMap.values())` 来实现这一点。此时，我们将记录错误并返回。

否则，如果一切成功，我们现在可以遍历 `flowFileMap` 并将每个 FlowFile 传输到相应的 Relationship。然后移除原始 FlowFile 或将其路由到 `original` Relationship。对于每个新创建的 FlowFile，我们还发出一个 Provenance ROUTE 事件，指示 FlowFile 路由到了哪个 Relationship。在 ROUTE 事件的详细信息中包含此 FlowFile 中包含多少条信息也很有帮助。这使数据流管理员在查看溯源血缘视图时能够轻松看到给定输入 FlowFile 有多少条信息路由到了每个 Relationship。

此外，某些处理器可能需要对发送到每个 Relationship 的数据进行“分组”，以便发送到同一 Relationship 的每个 FlowFile 具有相同的值。在我们的示例中，我们可能希望允许正则表达式具有捕获组，如果 CSV 中两个不同的行匹配正则表达式但捕获组的值不同，我们希望将它们添加到两个不同的 FlowFile 中。然后可以将匹配的值作为属性添加到每个 FlowFile 中。这可以通过将 `flowFileMap` 修改为 `Map<Relationship, Map<T, FlowFile>>` 来实现，其中 `T` 是分组函数的类型（在我们的示例中，Group 是一个 `String`，因为它是评估正则表达式捕获组的结果）。



=== 基于属性的路由

此处理器与上述基于内容路由处理器几乎相同。它也有两种不同的形式：一对一和一对多，就像基于内容路由处理器一样。然而，此处理器不会调用 ProcessSession 的 `read` 方法，因为它不读取 FlowFile 内容。此处理器通常非常快，因此 `@SupportsBatching` 注解在这种情况下可能非常重要。



=== 拆分内容（一对多）

此处理器通常不需要用户配置，除了要创建的每个拆分的大小。`onTrigger` 方法从其输入队列获取 FlowFile。创建一个 FlowFile 类型的 List。通过 ProcessSession 的 `read` 方法读取原始 FlowFile，并使用 InputStreamCallback。在 InputStreamCallback 中，读取内容直到达到应拆分 FlowFile 的位置。如果不需要拆分，回调返回，原始 FlowFile 路由到 `success`。在这种情况下，会发出一个 Provenance ROUTE 事件。通常，将 FlowFile 路由到 `success` 时不会发出 ROUTE 事件，因为这会生成非常冗长的血缘关系，难以导航。然而，在这种情况下，该事件很有用，因为我们通常会期望一个 FORK 事件，而没有任何事件可能会引起困惑。FlowFile 未被拆分而是传输到 `success` 的事实会被记录，然后方法返回。

如果达到需要拆分 FlowFile 的位置，则通过 ProcessSession 的 `create(FlowFile)` 方法或 `clone(FlowFile, long, long)` 方法创建新的 FlowFile。接下来的代码取决于使用的是 `create` 方法还是 `clone` 方法。两种方法都在下面描述。哪种方案合适必须根据具体情况确定。

当数据不会直接从原始 FlowFile 复制到新 FlowFile 时，Create 方法最合适。例如，如果只复制部分数据，或者数据在复制到新 FlowFile 之前需要以某种方式修改，则必须使用此方法。然而，如果新 FlowFile 的内容将是原始 FlowFile 某一部分的精确副本，则 Clone 方法更受青睐。

*Create 方法*
如果使用 `create` 方法，则以原始 FlowFile 作为参数调用该方法，以便新创建的 FlowFile 继承原始 FlowFile 的属性，并且框架会创建一个 Provenance FORK 事件。

然后代码进入 `try/finally` 块。在 `finally` 块中，新创建的 FlowFile 被添加到已创建的 FlowFile 列表中。这在 `finally` 块中完成，以便如果抛出异常，新创建的 FlowFile 将被正确清理。在 `try` 块中，回调通过调用 ProcessSession 的 `write` 方法并提供 OutputStreamCallback 来启动新的回调。然后将适当的数据从原始 FlowFile 的 InputStream 复制到新 FlowFile 的 OutputStream。

*Clone 方法*
如果新创建的 FlowFile 的内容只是原始 FlowFile 字节的一个连续子集，则优先使用 ProcessSession 的 `clone(FlowFile, long, long)` 方法而不是 `create(FlowFile)` 方法。在这种情况下，原始 FlowFile 中新 FlowFile 内容应开始的偏移量作为第二个参数传递给 `clone` 方法。新 FlowFile 的长度作为第三个参数传递给 `clone` 方法。例如，如果原始 FlowFile 为 10,000 字节，我们调用 `clone(flowFile, 500, 100)`，返回给我们的 FlowFile 在属性方面与 `flowFile` 相同。但是，新创建的 FlowFile 的内容长度为 100 字节，从原始 FlowFile 的偏移量 500 开始。也就是说，新创建的 FlowFile 的内容与你复制原始 FlowFile 的 500 到 599 字节相同。

clone 创建后，它被添加到 FlowFile 列表中。

在适用时，此方法比 Create 方法更受青睐，因为不需要磁盘 I/O。框架能够简单地创建一个引用原始 FlowFile 内容子集的新 FlowFile，而不是实际复制数据。然而，这并不总是可能的。例如，如果必须从原始 FlowFile 的开头复制头信息并添加到每个拆分的开头，则此方法不可行。


*两种方法*
无论使用 Clone 方法还是 Create 方法，以下都适用：

如果在 InputStreamCallback 中的任何时候达到处理无法继续的条件（例如，输入格式错误），应抛出 `ProcessException`。对 ProcessSession 的 `read` 方法的调用被包装在 `try/catch` 块中，捕获 `ProcessException`。如果捕获到异常，生成一条解释错误的日志消息。新创建的 FlowFile 列表通过 ProcessSession 的 `remove` 方法移除。原始 FlowFile 被路由到 `failure`。

如果没有出现问题，原始 FlowFile 被路由到 `original`，所有新创建的 FlowFile 被更新以包含以下属性：

[options="header"]
|===
| 属性名称 | 描述
| `split.parent.uuid` | 原始 FlowFile 的 UUID
| `split.index` | 表示此 FlowFile 在列表中的序号的递增数字（第一个 FlowFile
          创建的 FlowFile 值为 `0`，第二个值为 `1`，依此类推）
| `split.count` | 创建的拆分 FlowFile 总数
|===

新创建的 FlowFile 被路由到 `success`；此事件被记录；方法返回。


=== 基于内容更新属性

此处理器与上述基于内容路由的处理器非常相似。FlowFile 不是路由到 `matched` 或 `unmatched`，而是通常路由到 `success` 或 `failure`，并根据需要向 FlowFile 添加属性。要添加的属性以类似于基于内容路由（一对多）的方式配置，用户定义自己的属性。属性的名称指示要添加的属性的名称。属性的值指示要应用于数据的某些匹配标准。如果匹配标准与数据匹配，则添加一个名称与属性名称相同的属性。属性的值是内容中匹配的标准。

例如，评估 XPath 表达式的处理器可能允许输入用户定义的 XPath。如果 XPath 匹配 FlowFile 的内容，则该 FlowFile 将添加一个属性，名称等于属性名称，值等于匹配 XPath 的 XML 元素或属性的文本内容。`failure` Relationship 将在此示例中，如果传入的 FlowFile 不是有效的 XML，则使用 `failure` Relationship。无论是否找到匹配项，都会使用 `success` Relationship。然后可以在适当的时候使用此信息来路由 FlowFile。

此处理器发出一个 ATTRIBUTES_MODIFIED 类型的溯源事件。


=== 丰富/修改内容

丰富/修改内容模式非常常见且非常通用。此模式负责任何通用内容修改。在大多数情况下，此处理器标注了 `@SideEffectFree` 和 `@SupportsBatching` 注解。处理器具有任意数量的必需和可选属性，具体取决于处理器的功能。处理器通常具有 `success` 和 `failure` Relationship。当输入文件格式不符合预期时，通常使用 `failure` Relationship。

此处理器获取一个 FlowFile 并使用 ProcessSession 的 `write(StreamCallback)` 方法更新它，以便能够同时读取 FlowFile 的内容并写入 FlowFile 内容的下一个版本。如果在回调过程中遇到错误，回调将抛出 `ProcessException`。对 ProcessSession 的 `write` 方法的调用被包装在 `try/catch` 块中，捕获 `ProcessException` 并将 FlowFile 路由到 failure。

如果回调成功，则发出一个 CONTENT_MODIFIED 溯源事件。



== 错误处理

编写处理器时，可能会发生几种不同的意外情况。处理器开发者理解如果处理器不自行处理错误时 Lynxus 框架的行为机制非常重要，理解处理器需要执行什么样的错误处理也很重要。在这里，我们将讨论处理器应如何在工作过程中处理意外错误。


=== 处理器内的异常

在处理器的 `onTrigger` 方法执行期间，很多事情可能会出错。常见的失败条件包括：

  - 传入数据的格式不符合预期。
  - 与外部服务的网络连接失败。
  - 读取或写入磁盘数据失败。
  - 处理器或依赖库中存在 bug。

这些情况中的任何一种都可能导致从处理器抛出异常。从框架的角度来看，可以从处理器逃逸的异常有两种类型：`ProcessException` 和所有其他异常。

如果从处理器抛出 ProcessException，框架将假设这是一个已知的失败结果。此外，这是一种稍后尝试重新处理数据可能会成功的条件。因此，框架将回滚正在处理的会话并惩罚正在处理的 FlowFile。

然而，如果任何其他异常从处理器逃逸，框架将假设这是一个开发者未考虑到的失败。在这种情况下，框架也会回滚会话并惩罚 FlowFile。但是，在这种情况下，我们可能会遇到一些非常有问题的情况。例如，处理器可能处于不良状态，可能会持续运行，耗尽系统资源，而不提供任何有用的工作。例如，当持续抛出 NullPointerException 时，这种情况相当常见。为了避免这种情况，如果 ProcessException 以外的异常能够从处理器的 `onTrigger` 方法逃逸，框架还将“管理性让出”处理器。这意味着处理器在一段时间内不会被触发运行。时间量在 `lynxus.properties` 文件中配置，默认为 10 秒。


=== 回调内的异常：IOException、RuntimeException

大多数情况下，当处理器中发生异常时，它发生在回调内部（即 `InputStreamCallback`、`OutputStreamCallback` 或 `StreamCallback`）。也就是说，在处理FlowFile 的内容。回调允许抛出 `RuntimeException` 或 `IOException`。对于 RuntimeException，此异常将传播回 `onTrigger` 方法。对于 `IOException`，异常将被包装在 ProcessException 中，然后从框架抛出。

因此，建议使用回调的处理器在 `try/catch` 块中执行，并捕获 `ProcessException` 以及预期回调会抛出的任何其他 `RuntimeException`。但是，*不*建议处理器捕获一般的 `Exception` 或 `Throwable`。不鼓励这样做有两个原因。

首先，如果抛出了意外的 RuntimeException，这很可能是一个 bug，允许框架回滚会话将确保不会丢失数据，并确保数据流管理员能够通过将数据保留在原位排队中来按其认为合适的方式处理数据。

其次，当回调抛出 IOException 时，实际上有两种类型的 IOException：从处理器代码抛出的（例如，数据格式不符合预期或网络连接失败），以及从内容存储库（FlowFile 内容的存储位置）抛出的。如果是后者，框架将捕获此 IOException 并将其包装为 `FlowFileAccessException`，它继承自 `RuntimeException`。这样做是为了让异常逃逸 `onTrigger` 方法，使框架能够适当处理这种情况。捕获一般 Exception 会阻止这种情况发生。


=== 惩罚与让出

当处理过程中出现问题时，框架公开了两种方法来允许处理器开发者避免执行不必要的工作：“惩罚”和“让出”。这两个概念对于 Lynxus API 的新开发者来说可能会混淆。开发者可以通过调用 ProcessSession 的 `penalize(FlowFile)` 方法来惩罚 FlowFile。这会导致 FlowFile 本身在一段时间内对下游处理器不可访问。FlowFile 不可访问的时间由数据流管理员通过在处理器配置对话框中设置“惩罚持续时间”来确定。默认值为 30 秒。通常在处理器确定由于预计会自行解决的环境原因而无法处理数据时执行此操作。 一个很好的例子是 PutSFTP 处理器，如果 SFTP 服务器上已存在同名文件，它将惩罚 FlowFile。在这种情况下，处理器惩罚 FlowFile 并将其路由到 failure。然后数据流管理员可以将 failure 路由回同一个 PutSFTP 处理器。这样，如果存在同名文件，处理器在 30 秒内（或 DFM 为处理器配置的任何时间段）不会尝试再次发送文件。同时，它能够继续处理其他 FlowFile。

另一方面，让出允许处理器开发者向框架指示它在一段时间内无法执行任何有用的功能。这通常发生在与远程资源通信的处理器中。如果处理器无法连接到远程资源，或者远程资源预计会提供数据但报告没有数据，处理器应调用 `ProcessContext` 对象上的 `yield` 然后返回。通过这样做，处理器告诉框架不应浪费资源触发此处理器运行，因为它无能为力——最好使用这些资源让其他处理器运行。


=== 会话回滚

到目前为止，当我们讨论 `ProcessSession` 时，我们通常只是将其称为访问 FlowFile 的机制。然而，它提供了另一个非常重要的功能，即事务性。在 ProcessSession 上调用的所有方法都作为事务执行。当我们决定结束事务时，可以通过调用 `commit()` 或调用 `rollback()` 来实现。通常，这由 `AbstractProcessor` 类处理：如果 `onTrigger` 方法抛出异常，AbstractProcessor 将捕获异常，调用 `session.rollback()`，然后重新抛出异常。否则，AbstractProcessor 将在 ProcessSession 上调用 `commit()`。

然而，有时开发者会希望显式回滚会话。这可以随时通过调用 `rollback()` 或 `rollback(boolean)` 方法来实现。如果使用后者，布尔值指示从队列中提取的 FlowFile（通过 ProcessSession 的 `get` 方法）在添加回队列之前是否应被惩罚。

当调用 `rollback` 时，该会话中对 FlowFile 发生的任何修改都将被丢弃，这包括内容和属性修改。此外，所有溯源事件都将被回滚（通过为 `force` 参数传递 `true` 值发出的任何 SEND 事件除外）。从输入队列中提取的 FlowFile 然后被传输回输入队列（并可选择惩罚），以便可以再次处理它们。

另一方面，当调用 `commit` 方法时，FlowFile 的新状态被持久化在 FlowFile 仓库中，发生的任何溯源事件被持久化在溯源仓库中。之前的内容被销毁（除非另一个 FlowFile 引用了相同的内容），FlowFile 被传输到出站队列，以便下一个处理器可以操作数据。

同样重要的是要注意此行为如何受到使用 `cn.org.acal.lynxus.annotation.behavior.SupportsBatching` 注解的影响。如果处理器使用此注解，对 `ProcessSession.commit` 的调用可能不会立即生效。相反，这些提交可能被批量处理以提供更高的吞吐量。然而，如果在任何时候处理器回滚 ProcessSession，自上次调用 `commit` 以来的所有更改都将被丢弃，所有“批量”提交将生效。这些“批量”提交不会被回滚。




== 通用设计注意事项

设计处理器时，有一些重要的设计考虑需要牢记。开发者指南的这一部分提出了一些开发者在创建处理器时应该考虑的想法。

=== 考虑用户

开发处理器（或任何其他组件）时要牢记的最重要概念之一是你正在创建的用户体验。重要的是要记住，作为这样一个组件的开发者，你可能拥有其他人不具备的关于上下文的重要知识。应始终提供文档，以便不太熟悉该过程的人能够轻松使用它。

在考虑用户体验时，同样重要的是要注意一致性非常重要。最好坚持标准的<<naming-conventions>>。这适用于处理器名称、属性名称和值、Relationship 名称以及用户将体验到的任何其他方面。

简洁性至关重要！避免添加你不期望用户理解或更改的属性。作为开发者，我们被告知硬编码值是不好的。但这有时会导致开发者公开一些属性，当被要求澄清时，告诉用户只需保留默认值。这会导致混乱和复杂性。


=== 内聚性和可重用性

为了创建一个单一的内聚单元，开发者有时会将多个功能组合到一个处理器中。这对于处理器期望输入数据为 X 格式以便处理器可以将数据转换为 Y 格式并将新格式化的数据发送到某个外部服务的情况尤其如此。

采用这种为特定端点格式化数据然后在同一处理器中将数据发送到该端点的方法有几个缺点：

  - 处理器变得非常复杂，因为它必须执行数据转换任务以及将数据发送到远程服务的任务。
  - 如果处理器无法与远程服务通信，它将数据路由到 `failure` Relationship。在这种情况下，处理器将负责再次执行数据转换。如果再次失败，转换又要重新执行。
  - 如果我们有五个不同的处理器在发送数据之前将传入数据转换为这种新格式，我们会有大量重复的代码。例如，如果模式更改，许多处理器必须更新。
  - 当处理器完成发送到远程服务时，这些中间数据被丢弃。中间数据格式可能对其他处理器很有用。

为了避免这些问题，并使处理器更具可重用性，处理器应始终遵循“做一件事并做好”的原则。这样的处理器应拆分为两个独立的处理器：一个将数据从格式 X 转换为格式 Y，另一个将数据发送到远程资源。


[[naming-conventions]]
=== 命名约定

为了给用户提供一致的外观和体验，建议处理器遵循标准命名约定。以下是使用的标准约定列表：

  - 从远程系统拉取数据的处理器命名为 Get<Service> 或 Get<Protocol>，具体取决于它们是通过已知协议从任意源轮询数据（如 GetHTTP 或 GetFTP）还是从已知服务拉取数据（如 GetKafka）
  - 向远程系统推送数据的处理器命名为 Put<Service> 或 Put<Protocol>。
  - Relationship 名称使用小写字母，并使用空格分隔单词。
  - 属性名称将重要单词首字母大写，就像书名一样。



=== 处理器行为注解

创建处理器时，开发者能够向框架提供有关如何最有效地使用处理器的提示。这是通过向处理器的类应用注解来完成的。可应用于处理器的注解存在于 `cn.org.acal.lynxus.annotation` 的三个子包中。`documentation` 子包中的注解用于向用户提供文档。`lifecycle` 子包中的注解指示框架应在处理器上调用哪些方法以响应相应的生命周期事件。`behavior` 包中的注解帮助框架理解在调度和一般行为方面如何与处理器交互。

来自 `cn.org.acal.lynxus.annotation.behavior` 包的以下注解可用于修改框架处理处理器的方式：

  - `SideEffectFree`：表示处理器没有任何 Lynxus 外部的副作用。因此，框架可以自由地多次使用相同的输入调用处理器而不会导致任何意外结果。这意味着幂等行为。框架可以使用此注解通过执行诸如将 ProcessSession 从一个处理器传输到另一个处理器等操作来提高效率，这样如果
    出现问题时，许多处理器的操作可以回滚并重新执行。

  - `SupportsBatching`：此注解表示框架可以将多个 ProcessSession 提交批量合并为单个提交。如果存在此注解，用户将能够在处理器的调度选项卡中选择是偏好高吞吐量还是低延迟。此注解应应用于大多数处理器，但有一个注意事项：如果处理器调用 `ProcessSession.commit`，则无法保证数据安全地存储在 Lynxus 的内容、FlowFile 和溯源仓库中。因此，它不适用于那些从外部源接收数据、提交会话然后删除远程数据或与远程资源确认事务的处理器。

  - `TriggerSerially`：当存在此注解时，框架不允许用户调度多个并发线程同时执行 `onTrigger` 方法。相反，线程数（“并发任务”）将始终设置为 `1`。然而，这*并不*意味着处理器不必是线程安全的，因为执行 `onTrigger` 的线程可能在调用之间更改。

    - `PrimaryNodeOnly`：ACAL Lynxus 在集群模式下为处理器提供两种执行模式：“主节点”和“所有节点”。虽然在所有节点中运行提供了更好的并行性，但已知某些处理器在多个节点中运行时会导致意外行为。例如，某些处理器从远程文件系统列出或读取文件。如果将此类处理器调度为在“所有节点”上运行，将导致不必要的重复甚至错误。此类处理器应使用此注解。应用此注解将限制处理器仅在“主节点”上运行。

  - `TriggerWhenAnyDestinationAvailable`：默认情况下，如果任何出站队列已满，Lynxus 不会调度处理器运行。这允许在整个处理器链中应用反压。然而，某些处理器可能需要在某个出站队列已满时仍然运行。此注解表示如果任何 Relationship “可用”，处理器就应运行。如果使用某个 Relationship 的连接都没有满，则该 Relationship 被称为“可用”。例如，DistributeLoad 处理器使用了此注解。如果使用“轮询”调度策略，任何出站队列已满时处理器不会运行。但是，如果使用“下一个可用”调度策略，只要有任何 Relationship 可用，处理器就会运行，并仅将 FlowFile 路由到那些可用的 Relationship。

  - `TriggerWhenEmpty`：默认行为是仅在其输入队列中至少有一个 FlowFile 或处理器没有输入队列（这是“源”处理器的典型情况）时才触发处理器运行。应用此注解将导致框架忽略输入队列的大小，无论输入队列上是否有任何数据都触发处理器。例如，如果处理器需要定期触发运行以使网络连接超时，这很有用。

    - `InputRequirement`：默认情况下，所有处理器都允许用户为处理器创建传入连接，但如果用户没有创建传入连接，处理器仍然有效并且可以被调度运行。然而，对于预期作为“源处理器”使用的处理器，这可能会让用户困惑，用户可能会尝试将 FlowFile 发送到该处理器，但 FlowFile 只会排队而不被处理。相反，如果处理器期望传入的 FlowFile 但没有输入队列，处理器将被调度运行但不会执行任何工作，因为它不会收到任何 FlowFile，这同样会导致困惑。因此，我们可以使用 `@InputRequirement` 注解并提供 `INPUT_REQUIRED`、`INPUT_ALLOWED` 或 `INPUT_FORBIDDEN` 的值。这向框架提供关于何时应使处理器无效，或者用户是否应该能够绘制连接到该处理器的信息。例如，如果处理器标注了 `InputRequirement(Requirement.INPUT_FORBIDDEN)`，则用户甚至无法以该处理器为目标创建连接。


=== 数据缓冲

需要牢记的一点是，Lynxus 提供了通用的数据处理能力。数据可以是任何格式。处理器通常以多个线程调度。Lynxus 新手开发者常犯的一个错误是将 FlowFile 的所有内容缓冲到内存中。虽然有时需要这样做，但应尽可能避免，除非很清楚数据的格式。例如，负责对 XML 文档执行 XPath 的处理器需要将数据的全部内容加载到内存中。这通常是可以接受的，因为 XML 预计不会非常大。然而，搜索特定字节序列的处理器可能用于搜索数百 GB 或更大的文件。尝试将其加载到内存中可能会导致很多问题——特别是如果多个线程同时处理不同的 FlowFile。

不要将这些数据缓冲到内存中，建议在数据从内容仓库流式传输时进行评估（即，从 `ProcessSession.read` 提供给回调的 `InputStream` 中扫描内容）。当然，在这种情况下，我们不想为每个字节都从内容仓库读取，所以我们会使用 BufferedInputStream 或以某种方式缓冲少量数据。





[[controller-services]]
== 控制器服务

`ControllerService` 接口允许开发者以干净一致的方式在 JVM 中共享功能和状态。该接口类似于 `Processor` 接口，但没有 `onTrigger` 方法，因为控制器服务不被调度定期运行，而且控制器服务没有 Relationship，因为它们不直接集成到数据流中。相反，它们被处理器、报告任务和其他控制器服务使用。

[[developing-controller-service]]
=== 开发控制器服务

与处理器接口一样，ControllerService 接口公开了配置、验证和初始化的方法。这些方法都与处理器接口中的相同，只是 `initialize` 方法传递的是 `ControllerServiceInitializationContext` 而不是 `ProcessorInitializationContext`。

控制器服务有一个处理器所没有的额外约束，即控制器服务必须由一个扩展 `ControllerService` 的接口组成。然后只能通过其接口与实现进行交互。例如，处理器永远不会被给予 ControllerService 的具体实现，因此必须仅通过扩展 `ControllerService` 的接口引用服务。

这个约束存在的主要原因是处理器可以存在于一个 Lynxus 归档（NAR）中，而处理器所在的控制器服务的实现可以存在于不同的 NAR 中。这是由框架通过动态实现公开的接口来完成的，使框架可以切换到适当的 ClassLoader 并在具体实现上调用所需的方法。然而，为了使这项工作有效，处理器和控制器服务实现必须共享控制器服务接口的相同定义。因此，这两个 NAR 都必须依赖于包含控制器服务接口的 NAR。有关更多信息，请参阅<<nars>>。


[[interacting-with-controller-service]]
=== 与控制器服务交互

ControllerService 可以通过处理器、另一个 ControllerService、ParameterProvider 或 ReportingTask 通过 ControllerServiceLookup 或使用 PropertyDescriptor 的 Builder 类的 `identifiesControllerService` 方法来获取。ControllerServiceLookup 可以由处理器从传递给 `initialize` 方法的 ProcessorInitializationContext 中获取。同样，ControllerService 从 ControllerServiceInitializationContext 获取它，ParameterProvider 从 ParameterProviderInitializationContext 获取它，ReportingTask 通过传递给 `initialize` 方法的 ReportingConfiguration 对象获取它。

然而，对于大多数用例，使用 PropertyDescriptor Builder 的 `identifiesControllerService` 方法是首选的，也是最简单的方法。为了使用此方法，我们创建一个引用 ControllerService 的 PropertyDescriptor，如下所示：

[source,java]
----
public static final PropertyDescriptor SSL_CONTEXT_PROVIDER = new PropertyDescriptor.Builder()
  .name("SSL Context Provider")
  .description("Specifies the service for securing connections encrypted with Transport Layer Security")
  .required(true)
  .identifiesControllerService(SSLContextProvider.class)
  .build();
----

使用此方法，将提示用户提供应使用的 SSL Context Provider。这是通过向用户提供一个下拉菜单来完成的，用户可以从该菜单中选择任何已配置的 `SSLContextProvider` 实例，无论其实现方式如何。

为了使用此服务，处理器可以使用如下代码：

[source,java]
----
final SSLContextProvider sslContextProvider = context.getProperty(SSL_CONTEXT_PROVIDER)
  .asControllerService(SSLContextProvider.class);
----

请注意，`SSLContextProvider` 是一个扩展 ControllerService 的接口。

== 报告任务

到目前为止，我们几乎没有提到如何向外界传达 Lynxus 及其组件的性能状况。系统能否跟上传入数据的速率？系统还能处理多少数据？在一天中的高峰时段与最空闲时段分别处理了多少数据？

为了回答这些问题以及更多问题，Lynxus 提供了报告状态、统计信息、指标和监控信息到外部服务，通过 `ReportingTask` 接口。ReportingTask 可以访问大量信息以确定系统的运行状况。


=== 开发报告任务

与 Processor 和 ControllerService 接口一样，ReportingTask 接口公开了配置、验证和初始化的方法。这些方法与 Processor 和 ControllerService 接口中的方法完全相同，只是 `initialize` 方法传递的是 `ReportingInitializationContext` 对象，而不是其他组件接收的初始化对象。ReportingTask 还有一个 `onTrigger` 方法，由框架调用以触发任务执行其工作。

在 `onTrigger` 方法中，ReportingTask 可以访问 ReportingContext，从中可以获取配置和关于 Lynxus 实例的信息。BulletinRepository 允许查询 Bulletin 并允许 ReportingTask 提交自己的 Bulletin，以便将信息呈现给用户。通过 Context 可访问的 ControllerServiceLookup 提供对已配置的 ControllerService 的访问。然而，这种获取控制器服务的方法不是首选方法。首选方法是在 PropertyDescriptor 中引用控制器服务，如<<interacting-with-controller-service>>节所述。

通过 ReportingContext 公开的 `EventAccess` 对象提供对 `ProcessGroupStatus` 的访问，其中公开了流程组、处理器、连接和其他组件在过去五分钟内处理的数据量的统计信息。此外，EventAccess 对象提供对存储在 `ProvenanceEventRepository` 中的 ++ProvenanceEventRecord++ 的访问。这些溯源事件由处理器在从外部源接收数据、发送到外部服务、从系统中移除、修改或根据某个决策进行路由时发出。

每个 ProvenanceEvent 都有 FlowFile 的 ID、事件类型、事件创建时间，以及在组件访问 FlowFile 时与 FlowFile 关联的所有 FlowFile 属性，还有由于事件描述的处理而与 FlowFile 关联的 FlowFile 属性。这为 ReportingTask 提供了大量信息，允许以多种不同方式生成报告，以暴露任何数量的运营关注点所需的指标和监控功能。

== 参数提供程序

虽然参数上下文不能扩展，但 `ParameterProvider` 接口提供了一个扩展点，用于向参数上下文提供参数。参数提供程序可以获取参数组，这些参数组可以映射到新的参数上下文，然后通过重新获取参数来保持最新。

这种可扩展性允许参数在 Lynxus 外部进行初始提供和管理。


=== 开发参数提供程序

ParameterProvider 接口公开了配置、验证和初始化的方法。这些方法与 Processor 和 ControllerService 接口中的方法完全相同，只是 `initialize` 方法传递的是 `ParameterProviderInitializationContext` 对象，而不是其他组件接收的初始化对象。ParameterProvider 还有一个 `fetchParameters` 方法，由框架调用以从其源获取参数。

获取的参数在 `ParameterGroup` 中返回，每个参数组有一个组名和一个参数列表。用户可以将参数组映射到参数上下文。这允许单个 `ParameterProvider` 实例根据其映射向多个参数上下文提供不同的参数。

在 `fetchParameters` 方法中，ParameterProvider 可以访问 `ConfigurationContext`，从中可以获取配置和关于 Lynxus 实例的信息。通过 Context 可访问的 ControllerServiceLookup 提供对已配置的 ControllerService 的访问。然而，这种获取控制器服务的方法不是首选方法。首选方法是在 PropertyDescriptor 中引用控制器服务，如<<interacting-with-controller-service>>节所述。

== UI 扩展

Lynxus 中有两个 UI 扩展点可用：

- 自定义处理器 UI
- 内容查看器

可以创建自定义 UI 来提供超出大多数处理器设置中可用的标准属性/值表的配置选项。具有自定义 UI 的处理器示例包括 link:https://git.acal.org.cn/acal/lynxus/tree/main/lynxus-nar-bundles/lynxus-update-attribute-bundle[UpdateAttribute^] 和 link:https://git.acal.org.cn/acal/lynxus/tree/main/lynxus-nar-bundles/lynxus-standard-bundle[JoltTransformJSON^]。

可以创建内容查看器来扩展可在 Lynxus 中查看的数据类型。Lynxus 在 lib 目录中附带了包含内容查看器的 NAR，用于 csv、xml、avro、json（standard-nar）等数据类型以及 png、jpeg 和 gif（media-nar）等图像类型。

=== 自定义处理器 UI

要向处理器添加自定义 UI：

1. 创建你的 UI。2. 将 WAR 构建并打包到处理器 NAR 中。3. WAR 需要在 META-INF 目录中包含 `lynxus-processor-configuration` 文件，该文件将自定义 UI 与该处理器关联。4. 将 NAR 放在 lib 目录中，Lynxus 启动时会自动发现它。5. 在处理器的配置处理器窗口中，属性选项卡现在应有一个“高级”按钮，可以访问自定义 UI。

例如，以下是 UpdateAttribute 的 NAR 布局：

.UpdateAttribute NAR 布局
------------------------------------------------------------------------------------------------
lynxus-update-attribute-bundle
│
├── lynxus-update-attribute-model
│
├── lynxus-update-attribute-nar
│
├── lynxus-update-attribute-processor
│
├── lynxus-update-attribute-ui
│   ├── pom.xml
│   └── src
│       └── main
│           ├── java
│           ├── resources
│           └── webapp
│               └── css
│               └── images
│               └── js
│               └── META-INF
│               │   └── lynxus-processor-configuration
│               └── WEB-INF
│
└── pom.xml
------------------------------------------------------------------------------------------------

`lynxus-processor-configuration` 的内容如下：

`cn.org.acal.lynxus.processors.attributes.UpdateAttribute:${project.groupId}:lynxus-update-attribute-nar:${project.version}`

注意：自定义 UI 也可以为控制器服务和报告任务实现。


=== 内容查看器

添加内容查看器：

1. 将 WAR 构建并打包到处理器 NAR 中。2. WAR 需要在 META-INF 目录中包含 `lynxus-content-viewer` 文件，其中列出支持的内容类型。3. 将 NAR 放在 lib 目录中，Lynxus 启动时会自动发现它。4. 当遇到匹配的内容类型时，内容查看器将生成适当的视图。

一个可以参考的好示例是标准内容查看器的 NAR 布局：

.标准内容查看器 NAR 布局
------------------------------------------------------------------------------------------------
lynxus-standard-bundle
│
├── lynxus-jolt-transform-json-ui
│
├── lynxus-standard-content-viewer
│   ├── pom.xml
│   └── src
│       └── main
│           ├── java
│           ├── resources
│           └── webapp
│               └── css
│               └── META-INF
│               │   └── lynxus-content-viewer
│               └── WEB-INF
│
├── lynxus-standard-nar
│
├── lynxus-standard-prioritizers
│
├── lynxus-standard-processors
│
├── lynxus-standard-reporting-tasks
│
├── lynxus-standard-utils
│
└── pom.xml
------------------------------------------------------------------------------------------------

`lynxus-content-viewer` 的内容如下：
------------------------------------------------------------------------------------------------
application/xml
application/json
text/plain
text/csv
avro/binary
application/avro-binary
application/avro+binary
------------------------------------------------------------------------------------------------


== 测试

测试将在更大框架中使用的组件通常可能非常繁琐和困难。在 Lynxus 中，我们努力使组件测试尽可能简单。为此，我们创建了一个 `lynxus-mock` 模块，可以与 JUnit 结合使用来提供全面的组件测试。

Mock 框架主要针对测试处理器，因为这是迄今为止最常见的扩展点。但是，该框架确实提供了测试控制器服务的能力。

组件通常通过创建功能测试来测试，以验证组件行为。这样做是因为处理器通常由少量辅助方法组成，但逻辑主要在 `onTrigger` 方法中。`TestRunner` 接口允许我们通过将更“基础”的对象（如文件和字节数组）转换为 FlowFile 来测试处理器和控制器服务，并处理创建处理器工作所需的 ProcessSession 和 ProcessContext，以及调用必要的生命周期方法，以确保处理器在单元测试中的行为与在生产环境中相同。


=== 实例化 TestRunner

处理器或控制器服务的大多数单元测试首先创建 `TestRunner` 类的实例。为了将必要的类添加到处理器中，可以使用 Maven 依赖：

[source]
----
<dependency>
  <groupId>cn.org.acal.lynxus</groupId>
  <artifactId>lynxus-mock</artifactId>
  <version>${lynxus version}</version>
</dependency>
----

我们通过调用 `TestRunners` 类（位于 `cn.org.acal.lynxus.util` 包中）的静态 `newTestRunner` 方法之一来创建新的 `TestRunner`。这些方法接受被测处理器的参数（可以是要测试的处理器类或处理器实例），并允许设置处理器名称。

=== 添加控制器服务

创建新的 Test Runner 后，我们可以将处理器使用的任何控制器服务添加到 Test Runner 中执行其工作所需的控制器服务。我们通过调用 `addControllerService` 方法并提供 ControllerService 的标识符和 ControllerService 的实例来实现这一点。

如果需要配置 ControllerService，可以通过调用 `setProperty(ControllerService, PropertyDescriptor, String)`、`setProperty(ControllerService, String, String)` 或 `setProperty(ControllerService, PropertyDescriptor, AllowableValue)` 方法来设置其属性。这些方法中的每一个都返回一个 `ValidationResult`。然后可以通过调用 `isValid` 来检查此对象以确保属性有效。可以通过调用 `setAnnotationData(ControllerService, String)` 方法来设置注解数据。

我们现在可以通过调用 `assertValid(ControllerService)` 来确保 ControllerService 有效——或者如果在测试 ControllerService 本身，可以通过调用 `assertNotValid(ControllerService)` 来确保配置的值无效。

一旦 ControllerService 被添加到 Test Runner 并配置完成，现在可以通过调用 `enableControllerService(ControllerService)` 方法来启用它。如果 ControllerService 无效，此方法将抛出 IllegalStateException。否则，服务现在可以使用了。



=== 设置属性值

配置任何必要的控制器服务后，我们需要配置处理器。我们可以通过调用与控制器服务相同的方法来实现，不指定任何控制器服务。即，我们可以调用 `setProperty(PropertyDescriptor, String)` 等等。每个 `setProperty` 方法再次返回一个 `ValidationResult` 属性，可用于确保属性值有效。

同样，我们也可以调用 `assertValid()` 和 `assertNotValid()` 来根据我们的预期确保处理器的配置是否有效。


=== 入队 FlowFile

在触发处理器运行之前，通常需要为处理器入队要处理的 FlowFile。这可以通过使用 `TestRunner` 类的 `enqueue` 方法来实现。`enqueue` 方法有几种不同的重载，允许以 `byte[]`、`InputStream` 或 `Path` 的形式添加数据。这些方法中的每一个都支持一个变体，允许添加 `Map<String, String>` 以支持 FlowFile 属性。

此外，还有一个 `enqueue` 方法接受 FlowFile 对象的可变参数。例如，这对于获取处理器的输出然后将其馈送到处理器的输入很有用。


=== 运行处理器

配置控制器服务并入队必要的 FlowFile 后，可以通过调用 `TestRunner` 的 `run` 方法触发处理器运行。如果调用此方法时不带任何参数，它将调用处理器中标注了 `@OnScheduled` 的任何方法，调用处理器的 `onTrigger` 方法一次，然后运行 `@OnUnscheduled` 方法，最后运行 `@OnStopped` 方法。

如果希望在触发其他 `@OnUnscheduled` 和 `@OnStopped` 生命周期事件之前运行多次 `onTrigger` 方法迭代，可以使用 `run(int)` 方法指定应调用多少次 `onTrigger` 迭代。

有时我们想触发处理器运行但不触发 `@OnUnscheduled` 和 `@OnStopped` 生命周期事件。例如，这对于在这些事件发生之前检查处理器的状态很有用。这可以通过使用 `run(int, boolean)` 并传递 `false` 作为第二个参数来实现。但是，在这样做之后，调用 `@OnScheduled` 生命周期方法可能会导致问题。因此，我们现在可以通过使用 `run` 方法的 `run(int,boolean,boolean)` 版本并传递 `false` 作为第三个参数来再次运行 `onTrigger` 而不触发这些事件。

如果测试多线程发生的行为很有用，这也可以通过调用 `TestRunner` 的 `setThreadCount` 方法来实现。默认值为 1 个线程。如果使用多个线程，重要的是要记住 `TestRunner` 的 `run` 调用指定处理器应被触发的次数，而不是每个线程应触发处理器的次数。因此，如果线程数设置为 2 但调用了 `run(1)`，则只使用单个线程。


=== 验证输出

处理器运行完成后，单元测试通常要验证 FlowFile 是否去了预期的地方。这可以通过使用 `TestRunner` 的 `assertAllFlowFilesTransferred` 和 `assertTransferCount` 方法来实现。前一个方法接受一个 Relationship 和一个整数作为参数，以指定应传输到该 Relationship 的 FlowFile 数量。除非这个数量的 FlowFile 被传输到给定的 Relationship，或者任何 FlowFile 被传输到任何其他 Relationship，否则该方法将使单元测试失败。`assertTransferCount` 方法仅验证 FlowFile 计数是否是给定 Relationship 的预期数量。

验证计数后，我们可以通过 `getFlowFilesForRelationship` 方法获取实际的输出 FlowFile。此方法返回一个 `List<MockFlowFile>`。重要的是要注意 List 的类型是 `MockFlowFile`，而不是 `FlowFile` 接口。这样做是因为 `MockFlowFile` 附带了许多验证内容的方法。

例如，`MockFlowFile` 有用于断言 FlowFile 属性存在（`assertAttributeExists`）、断言其他属性不存在（`assertAttributeNotExists`）或属性具有正确值（`assertAttributeEquals`、`assertAttributeNotEquals`）的方法。还有类似的方法用于验证 FlowFile 的内容。FlowFile 的内容可以与 `byte[]`、`InputStream`、文件或 String 进行比较。如果数据预期为文本，则首选 String 版本，因为如果输出不符合预期，它提供更直观的错误消息。



=== 模拟外部资源

测试连接到远程资源的 Lynxus 处理器时最大的问题之一是我们不想从单元测试中实际连接到某个远程资源。我们可以在单元测试中自己搭建一个简单的服务器并配置处理器与之通信，但这需要我们理解和实现特定于服务器的规范，并且可能无法正确发回我们想要用于测试的错误消息等。

通常，这里采取的方法是在处理器中有一个负责获取到远程资源的连接或客户端的方法。我们通常将此方法标记为 protected。在单元测试中，我们不是通过调用 `TestRunners.newTestRunner(Class)` 并提供处理器类来创建 `TestRunner`，而是在单元测试中创建处理器的子类并使用它：

[source,java]
----
@Test
public void testConnectionFailure() {
  final TestRunner runner = TestRunners.newTestRunner(new MyProcessor() {
    protected Client getClient() {
      // Return a mocked out client here.
      return new Client() {
        public void connect() throws IOException {
          throw new IOException();
        }

        // ...
        // other client methods
        // ...
      };
    }
  });

  // rest of unit test.
}
----

这允许我们实现一个模拟所有网络通信的 Client，返回我们想要测试的不同错误结果，并确保我们的逻辑正确处理对客户端的成功调用。


=== 其他测试功能

除了测试框架提供的上述功能外，TestRunner 还提供了几个便捷方法来验证处理器的行为。提供了确保处理器的输入队列已清空的方法。单元测试能够获取 TestRunner 将使用的 ProcessContext、ProcessSessionFactory、ProvenanceReporter 和其他框架特定的实体。`shutdown` 方法提供了测试标注为仅在 Lynxus 关闭时运行的处理器方法的能力。可以为使用自定义用户界面的处理器设置注解数据。最后，可以通过 `setThreadCount(int)` 方法设置应用于运行处理器的线程数。






[[nars]]
== Lynxus 归档 (NAR)

当来自许多不同组织的软件都托管在同一环境中时，Java ClassLoader 很快就会成为一个问题。如果多个组件依赖同一个库但每个依赖不同的版本，会出现许多问题，通常导致意外行为或 `NoClassDefFoundError` 错误。为了防止这些问题变得棘手，Lynxus 引入了 Lynxus 归档（NAR）的概念。

NAR 允许将多个组件及其依赖项打包到一个单独的包中。然后 NAR 包提供与其他 NAR 包的 ClassLoader 隔离。开发者应始终将 Lynxus 组件部署为 NAR 包。

为了实现这一点，开发者创建一个新的 Maven 构件，我们称之为 NAR 构件。打包类型设置为 `nar`。然后创建 POM 的 `dependencies` 部分，使 NAR 依赖于要包含在 NAR 中的所有 Lynxus 组件。

为了使用 `nar` 打包类型，我们必须使用 `lynxus-nar-maven-plugin` 模块。这通过在 NAR 的 pom.xml 中添加以下代码片段来包含：

[source,xml]
----
<build>
    <plugins>
        <plugin>
            <groupId>cn.org.acal.lynxus</groupId>
            <artifactId>lynxus-nar-maven-plugin</artifactId>
            <version>1.3.4</version>
            <extensions>true</extensions>
        </plugin>
    </plugins>
</build>
----

在 ACAL Lynxus 代码库中，这存在于 Lynxus 根 POM 中，所有其他 Lynxus 构件（lynxus-nar-maven-plugin 本身除外）都继承自它，因此我们不需要在任何其他 POM 文件中包含它。

NAR 能够有一个类型为 `nar` 的依赖项。如果指定了多个类型为 `nar` 的依赖项，则 lynxus-nar-maven-plugin 将报错。如果 NAR A 添加了对 NAR B 的依赖，这*不会*导致 NAR B 打包 NAR A 的所有组件。相反，这将在 `MANIFEST.MF` 中添加一个 `Nar-Dependency-Id` 元素文件的 NAR A。这将导致将 NAR B 的 ClassLoader 设置为 NAR A 的父 ClassLoader。在这种情况下，我们将 NAR B 称为 NAR A 的 _父_ 级。

这种父 ClassLoader 的链接是 Lynxus 用来使控制器服务能够在所有 NAR 之间共享的机制。如<<developing-controller-service>>节所述，控制器服务必须分为一个扩展 `ControllerService` 的接口和一个实现该接口的实现。控制器服务可以从任何处理器引用，无论它在哪个 NAR 中，只要控制器服务实现和处理器共享控制器服务接口的相同定义。

为了共享这个相同的定义，处理器的 NAR 和控制器服务实现的 NAR 都必须以控制器服务定义的 NAR 作为父级。示例层次结构可能如下所示：

.控制器服务 NAR 布局
------------------------------------------------------------------------------------------------
root
├── my-controller-service-api
│   ├── pom.xml
│   └── src
│       └── main
│           └── java
│               └── org
│                   └── my
│                       └── services
│                           └── MyService.java
│
├── my-controller-service-api-nar
│   └── pom.xml (1)
│
│
│
├── my-controller-service-impl
│   ├── pom.xml (2)
│   └── src
│       ├── main
│       │   ├── java
│       │   │   └── org
│       │   │       └── my
│       │   │           └── services
│       │   │                 └── MyServiceImpl.java
│       │   └── resources
│       │       └── META-INF
│       │           └── services
│       │               └── cn.org.acal.lynxus.controller.ControllerService
│       └── test
│           └── java
│               └── org
│                   └── my
│                       └── services
│                             └── TestMyServiceImpl.java
│
│
├── my-controller-service-nar
│   └── pom.xml (3)
│
│
└── other-processor-nar
    └── pom.xml (3)




------------------------------------------------------------------------------------------------
<1> 此 POM 文件的类型为 `nar`。它依赖于 `lynxus-standard-services-api-nar`。<2> 此 POM 文件的类型为 `jar`。它依赖于 `my-controller-service-api`。它*不*依赖于任何 `nar` 构件。<3> 此 POM 文件的类型为 `nar`。它依赖于 `my-controller-service-api-nar`。


虽然这些在开始时可能看起来非常复杂，但在创建一两次这样的层次结构后，它变得远不那么复杂。请注意，`my-controller-service-api-nar` 依赖于 `lynxus-standard-services-api-nar`。这样做是为了使任何依赖于 `my-controller-service-api-nar` 的 NAR 也能够访问 `lynxus-standard-services-api-nar` 提供的所有控制器服务，例如 `SSLContextProvider`。同样，不需要为每个服务创建不同的“service-api” NAR。相反，通常有一个封装许多不同控制器服务 API 的单一“service-api” NAR 是有意义的，就像 `lynxus-standard-services-api-nar` 所做的那样。通常，API 不包括大量依赖项，因此 ClassLoader 隔离可能不那么重要，所以将许多 API 构件合并到同一个 NAR 中通常是可以接受的。




[[per-instance-classloading]]
== 每实例类加载

组件开发者可能希望在运行时向组件的类路径添加额外资源。例如，你可能想要向与关系数据库交互的处理器提供 JDBC 驱动程序的位置，从而允许处理器与任何驱动程序一起工作，而不是尝试将驱动程序打包到 NAR 中。

这可以通过声明一个或多个 `dynamicallyModifiesClasspath` 设置为 true 的 PropertyDescriptor 实例来实现。例如：


[source,java]
----
PropertyDescriptor EXTRA_RESOURCE = new PropertyDescriptor.Builder()
   .name("Extra Resources")
   .description("The path to one or more resources to add to the classpath.")
   .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
   .expressionLanguageSupported(true)
   .dynamicallyModifiesClasspath(true)
   .build();
----

当在组件上设置这些属性时，框架识别所有 `dynamicallyModifiesClasspath` 设置为 true 的属性。对于这些属性中的每一个，框架尝试从属性值解析文件系统资源。值可以是一个或多个目录或文件的逗号分隔列表，其中不存在的路径将被跳过。如果资源表示一个目录，则列出该目录，并将目录中的所有文件单独添加到类路径中。这些目录还将被扫描以查找本机库。如果在这些目录之一中找到库，则在加载之前创建并缓存由操作系统处理的临时副本，以保持一致性和类加载器隔离。

每个属性可能通过验证器对值的格式施加进一步的限制。例如，使用 `StandardValidators.FILE_EXISTS_VALIDATOR` 将属性限制为接受单个文件。使用 `StandardValidators.NON_EMPTY_VALIDATOR` 允许逗号分隔的文件或目录的任何组合。

通过将资源添加到始终首先检查的内部 ClassLoader 来将资源添加到实例 ClassLoader。每当这些属性的值更改时，内部 ClassLoader 将被关闭并用新资源重新创建。

Lynxus 提供了 `@RequiresInstanceClassLoading` 注解来进一步扩展和隔离组件类路径上可用的库。你可以使用 `@RequiresInstanceClassLoading` 注解标注组件，以指示组件的实例 ClassLoader 需要组件 NAR ClassLoader 中所有资源的副本。当不存在 `@RequiresInstanceClassLoading` 时，实例 ClassLoader 只是将其父 ClassLoader 设置为 NAR ClassLoader，而不是复制资源。

`@RequiresInstanceClassLoading` 注解还提供了一个可选标志 `cloneAncestorResources`。如果设置为 true，实例 ClassLoader 将包含祖先资源，直到包含组件引用的控制器服务 API 的第一个 ClassLoader，或直到 Jetty NAR。如果设置为 false 或未指定，则只包含组件 NAR 中的资源。

由于 `@RequiresInstanceClassLoading` 为组件的每个实例从 NAR ClassLoader 复制资源，因此应谨慎使用此功能。如果创建了一个组件的十个实例，则组件 NAR ClassLoader 中的所有类都会被加载到内存中十次。当创建足够多的组件实例时，这最终可能会显著增加内存占用。

此外，在使用 ControllerService 时使用 `@RequiresInstanceClassLoading` 存在一些限制。处理器、报告任务和 ControllerService 可以在其 PropertyDescriptor 之一中引用 ControllerService API。当 ControllerService API 与引用它的组件或 ControllerService 实现打包在同一个 NAR 中时，可能会出现问题。如果遇到这两种情况之一并且扩展需要实例类加载，则扩展将被跳过并记录适当的 ERROR。要解决此问题，ControllerService API 应打包在父 NAR 中。服务实现和引用该服务的扩展应依赖于 ControllerService API NAR。请参阅<<nars>>节中的 ControllerService NAR 布局。每当 ControllerService API 与需要它的扩展打包在一起时，即使没有使用 `@RequiresInstanceClassLoading`，也会记录 WARNING 以帮助避免这种不良做法。


[[deprecation]]
== 废弃组件和功能
废弃功能是软件开发生命周期的重要组成部分，为开发者和用户提供升级路径。

ACAL Lynxus 遵循 link:https://semver.org[语义化版本规范 2.0.0^]，用于根据 link:https://cwiki.apache.org/confluence/display/NIFI/Version+Scheme+and+API+Compatibility[版本方案] 文档中定义的公共版本合约所标识的功能。

属于公共版本合约范围内的组件和功能在移除之前需要进行废弃标记。这种方法允许开发者在次要版本升级时实施更改，为后续主要版本中移除功能做准备。

=== 组件废弃

有时可能需要废弃一个组件。发生这种情况时，开发者可以使用 `@DeprecationNotice` 注解来指示组件已被废弃，允许开发者描述废弃的 `reason`（原因）并建议替代组件。以下是如何执行此操作的示例：

[source, java]
----
@DeprecationNotice(
        reason = "Legacy Protocol has been superseded",
        alternatives = {SomeNewProcessor.class},
        classNames = {"cn.org.acal.lynxus.processors.standard.SomeOldProcessor"}
)
public class SomeOldProcessor extends AbstractProcessor {}
----
`alternatives` 属性可用于定义推荐替换组件的数组，而 `classNames` 可用于通过类名字符串数组表示类似的内容。

添加 `@DeprecationNotice` 注解会在生成的文档中呈现警告消息，并在流配置包含该组件时记录以下警告：

----
Added Deprecated Component SomeOldProcessor[id=929a52c7-1e3e-423e-b303-6ca2ef657617] See alternatives [SomeNewProcessor]
----

=== 功能废弃

废弃功能包括更改组件配置策略、引入新的仓库类和重构控制器服务接口。移除组件属性可能在升级后创建无效的流配置，从控制器服务接口移除公共方法可能会破坏针对先前版本编译的组件。由于这些原因，引入新属性和方法必须包含支持从一个次要版本升级到另一个次要版本时的兼容性的废弃策略。

使用 Java `@Deprecated` 注解标注方法和属性可以为软件开发者提供警告，但不向用户提供任何信息。添加选择性废弃日志记录为用户提供了一种确定当前流配置是否使用废弃属性或功能的方法。

=== 废弃日志记录

`lynxus-deprecation-log` 模块提供了记录废弃警告的抽象。`DeprecationLogger` 接口有一个 `warn()` 方法，接受带有可变数量参数的消息字符串。标准实现委托给 SLF4J，支持提供和解析占位符参数的标准功能。标准 `DeprecationLogger` 实现还记录带有引用类的 `DeprecationException`，支持追踪到使用位置。

实例化 `DeprecationLogger` 遵循与实例化 SLF4J `Logger` 相同的约定：

[source, java]
----
private static final DeprecationLogger deprecationLogger = DeprecationLoggerFactory.getLogger(Service.class);
----

记录废弃扩展组件属性和方法的警告应包括简单类名和组件标识符以便追踪。警告消息还应引用推荐的替代方案。

[source, java]
----
deprecationLogger.warn(
        "{}[id={}] createServiceContext() should be replaced with createContext()",
        getClass().getSimpleName(),
        getIdentifier()
);
----

[[build]]
== 构建选项
=== Maven 配置文件

[options="header,footer"]
|==================================================================================================================================================
| 包                        | Maven 配置文件            | 描述
| Apache Hadoop Bundle   | include-hadoop        | 添加对 Apache Hadoop 的支持，包括 HDFS 和 Parquet 组件
| Apache IoTDB Bundle    | include-iotdb         | 添加对 Apache IoTDB 的支持
| ASN.1 Support          | include-asn1          | 添加对 ASN.1 的支持
| Contribution Check     | contrib-check         | 运行在贡献被接受到核心 Lynxus 代码库之前需要的各种质量检查。
| Graph Database Bundle  | include-graph         | 添加对各种常见图数据库场景的支持。目前支持 https://neo4j.com/developer/cypher[Cypher] 和 https://tinkerpop.apache.org/gremlin.html[Gremlin] 兼容数据库，如 Neo4J 和 JanusGraph。包括提供驱动程序功能的控制器服务和一套用于摄入和查询的处理器。
| Media Bundle           | include-media         | 媒体包提供基于 https://tika.apache.org[Apache Tika] 的功能，用于从 Apache Tika 支持的各种二进制格式（如 PDF、Microsoft Office）中提取内容和元数据。
| QuestDB Bundle         | include-questdb       | 添加使用 QuestDB 的持久状态历史支持
| Snowflake Bundle       | include-snowflake     | 添加与 https://www.snowflake.com[Snowflake 平台]集成的支持。
| SQL Reporting Bundle   | include-sql-reporting | 添加设计为使用 SQL 用 ACAL Lynxus 的指标和其他相关数据更新 RDBMS 的报告任务。
|==================================================================================================================================================

=== 标准构建说明

以下命令用于生成 ACAL Lynxus 的标准二进制分发版：

`mvn clean install -Pcontrib-check`

== 支持从版本化进程组迁移

连接器可以实现一次性迁移流，使用源版本化流程组作为填充新创建的连接器的参考。这适用于能够将导出的流定义、其参数上下文、引用资产和任何附加组件状态转换为其自己的受管理流的连接器。源版本化流程组不安装在连接器上；相反，连接器读取源作为输入并更新自己的流以镜像源的配置和状态。

迁移支持是可选的。连接器通过实现可选功能接口 `cn.org.acal.lynxus.components.connector.migration.MigratableConnector` 来选择加入。未实现 `MigratableConnector` 的连接器永远不会被提供为迁移目标，并继续像以前一样运行。

=== 迁移契约

`MigratableConnector` 是一个独立的接口，连接器类必须实现该接口（通常与扩展 `AbstractConnector` 一起）。它公开了三个方法：

- `isMigrationSupported(ConnectorMigrationContext)` 是一个轻量级、无副作用的谓词。它应检查 `context.getSourceFlow()` 并仅在连接器可以从源结构迁移时返回 `true`。实现应将自己限制在结构检查中，如处理器类型、参数名称和导出的元数据。组件状态和资产内容在列出时不可用，不得依赖；每个状态的前提条件检查应放在 `migrateConfiguration` 或 `migrateState` 中。此方法不得调用 `ConnectorMigrationContext.copyAssetFromSource(...)` 或下面描述的任何写入 API；框架包装了资格检查时的上下文，因此任何此类尝试都会被 `IllegalStateException` 拒绝，连接器会从列表中过滤掉。
- `migrateConfiguration(ConnectorMigrationContext)` 记录连接器在迁移另一端想要的配置。连接器读取 `context.getSourceFlow()` 并通过 `context.setProperties(stepName, propertyValues)`（合并）或 `context.replaceProperties(stepName, propertyValues)`（完全替换步骤）记录配置更改。这些调用应用于框架用连接器当前活动配置播种的连接器配置的工作副本，因此工作副本始终反映完全合并的结果。`migrateConfiguration` 返回后，框架从该合并配置驱动 `Connector.applyUpdate(workingContext, activeContext)`，以便受管理的流程组由框架在重启时使用的相同代码路径重建。连接器不得从此方法调用 `updateFlow(...)`；框架向连接器传递活动 `FlowContext` 的只读包装器，因此任何直接的 `updateFlow(...)` 尝试都会被 `IllegalStateException` 拒绝。组件状态写入在此阶段也被拒绝：`setComponentState(...)` 会抛出异常，因为受管理的组件尚不存在。
- `migrateState(ConnectorMigrationContext)` 在框架从合并配置重建受管理的流程组之后运行。此时，源流中的每个处理器和控制器服务在活动 `FlowContext` 中都有相应的受管理组件，因此连接器可以通过版本化标识符将每个源 `VersionedConfigurableExtension` 与其受管理的对应组件匹配。连接器通过 `context.setComponentState(versionedComponentId, versionedComponentState)` 记录每个受管理组件所需的 `StateManager` 状态。`migrateState` 返回后，框架将每个暂存的 `VersionedComponentState` 写入相应受管理组件的活动 `StateManager`。由于上述相同原因，配置写入（`setProperties`、`replaceProperties`）和直接的 `updateFlow(...)` 调用在此阶段被拒绝。

这两个阶段共同形成一个单一的、全有或全无的迁移。只有在 `migrateState` 和框架后续的状态应用步骤都成功之后，框架才会将合并的配置提交到连接器的持久化活动配置中。任何一个阶段的失败都会回滚迁移：受管理的流程组从 `getInitialFlow()` 恢复，工作配置副本被丢弃，通过 `copyAssetFromSource(...)` 复制的任何资产被删除，连接器的持久化配置保持不变。

=== 访问源数据流

`ConnectorMigrationContext` 使用 `getSourceFlow()` 公开源 `VersionedExternalFlow`。源包括：

- 导出的 `VersionedProcessGroup`
- 参数上下文
- 引用资产
- 附加的 `@Stateful` 组件的 `VersionedComponentState`
- 导出元数据，如流程名称、注册表标识符和版本

同一上下文还使用 `isLocalMigration()` 指示请求是来自同一 Lynxus 实例上的本地流程组还是来自上传的有效负载。

活动 `FlowContext` 可通过 `context.getActiveFlowContext()` 获取，并且在迁移期间是只读的：它将读取（配置、受管理的流程组）转发到实时活动流上下文，但以 `IllegalStateException` 拒绝 `updateFlow(...)`。连接器应将此对象视为连接器当前状态的窗口，而不是写入目标。`migrateConfiguration` 返回后，框架通过调用 `Connector.applyUpdate(workingContext, activeContext)` 以相同的活动 `FlowContext` 作为目标来重建受管理的流程组，因此连接器针对 `getActiveFlowContext()` 解析的任何组件引用在应用迁移的流后仍然有效。

=== 记录配置更改

典型的 `migrateConfiguration` 实现将源流转换为连接器稍后可以从中重建的表示（通常是源 `VersionedExternalFlow` 的序列化版本或派生配置），然后将其暂存到连接器自己的配置上：

[source,java]
----
@Override
public void migrateConfiguration(final ConnectorMigrationContext context) throws FlowUpdateException {
    final VersionedExternalFlow sourceFlow = context.getSourceFlow();
    final String serializedSourceFlow = OBJECT_MAPPER.writeValueAsString(sourceFlow);

    // Record the configuration the Connector wants on the other side of the migration. The framework merges this
    // delta onto the active configuration and then drives Connector.applyUpdate(workingContext, activeContext),
    // which is the same path used on every restart - so the managed Process Group is rebuilt the same way after
    // migration and after every subsequent restart.
    context.setProperties(MIGRATION_STATE_STEP_NAME, Map.of(MIGRATED_SOURCE_FLOW_PROPERTY_NAME, serializedSourceFlow));
}
----

`setProperties` 是增量式的：步骤上的现有属性保持其值，除非暂存的映射覆盖它们。`replaceProperties` 是破坏性的：步骤被重置为提供的属性映射，映射中不存在的任何先前属性都将被移除。

连接器的 `applyUpdate(workingContext, activeContext)` 实现从提供的 `workingContext` 读取合并的配置，重建所需的受管理流程组，并调用 `getInitializationContext().updateFlow(activeContext, rebuiltFlow)` 来安装它。由于框架在重启期间通过 `inheritConfiguration(...)` 驱动相同的 `applyUpdate(...)` 调用，迁移的受管理流在每次重启时从持久化配置自动重建；连接器不需要做任何额外的工作来使迁移在重启后继续存在。

=== 记录组件状态

受管理的流程组重建后，`migrateState` 记录各个受管理组件的 `StateManager` 状态：

[source,java]
----
@Override
public void migrateState(final ConnectorMigrationContext context) {
    final VersionedProcessGroup sourceGroup = context.getSourceFlow().getFlowContents();
    for (final VersionedProcessor sourceProcessor : sourceGroup.getProcessors()) {
        final VersionedComponentState sourceState = sourceProcessor.getComponentState();
        if (sourceState != null) {
            context.setComponentState(sourceProcessor.getIdentifier(), sourceState);
        }
    }
}
----

`setComponentState` 的第一个参数是源端的版本化标识符；框架将其解析为匹配的受管理处理器或控制器服务，并将提供的 `VersionedComponentState` 写入该组件的活动 `StateManager`。集群范围的状态在每个节点上无条件应用；每个节点的本地范围状态由节点序号选择，因此集群迁移在每个节点上安装正确的条目。`@Stateful` 注解未声明请求范围的组件将被跳过并显示警告，而不是使迁移失败。

`setComponentState` 仅暂存所需状态；框架在 `migrateState` 返回后写入它。如果连接器对同一版本化标识符多次调用 `setComponentState`，则只有最后一次调用生效。

=== 重写参数和资产

导出的源流包含参数上下文和附加到导出参数的任何资产引用。连接器负责在 `migrateConfiguration` 中将这些值映射到自己的目标形状。

连接器可以使用 `ConnectorMigrationContext.copyAssetFromSource(String)` 将资产从本地源流程组复制到其自己的资产命名空间中。此方法仅适用于本地迁移。当提供的源资产标识符为 null 或空白时，它抛出 `IllegalArgumentException`；当为上传的有效负载迁移调用时，抛出 `IllegalStateException`。上传的有效负载无法访问活动的源资产命名空间，因此来自上传有效负载的资产引用必须在迁移请求之外处理。

以下示例定位源参数的引用资产，将其复制到连接器自己的资产命名空间中，并在目标属性上记录返回的 `AssetReference`：

[source,java]
----
final VersionedParameter sourceParameter = findSourceParameter(context.getSourceFlow(), sourceParameterName);
if (sourceParameter != null && sourceParameter.getReferencedAssets() != null && !sourceParameter.getReferencedAssets().isEmpty() && context.isLocalMigration()) {
    final VersionedAsset referencedAsset = sourceParameter.getReferencedAssets().getFirst();
    final AssetReference assetReference = context.copyAssetFromSource(referencedAsset.getIdentifier());
    context.setValueReference(targetStepName, targetPropertyName, assetReference);
}
----

=== 敏感值

敏感参数值和其他机密材料不存在于导出的源流中。连接器必须在迁移期间保持这些值未设置。迁移完成后，用户通过正常的配置步骤提供缺少的值，然后启动连接器。

=== 其他期望

- `isMigrationSupported(...)` 不得更改流状态、参数值或资产。
- `migrateConfiguration` 和 `migrateState` 应成功完成或抛出异常。任何抛出的异常都会中止迁移并回滚每个副作用：复制的资产被删除，受管理的流程组从 `getInitialFlow()` 恢复，每个组件的 `StateManager` 状态被清除，持久化的活动配置保持不变。
- 一旦连接器从其创建时的状态被修改，迁移将被拒绝。当活动或工作配置上的任何配置属性与该属性的声明默认值不同，或者受管理流中的任何处理器或控制器服务已存储组件状态时，框架认为连接器已被修改。
  此检查在调用 `migrateConfiguration` 时进行评估，并在每次评估 `MIGRATE` 允许操作时再次评估，因此用户可以看到迁移不可用的原因。此检查是重启安全的，因为持久化配置和任何存储的组件状态每次都从持久源重新计算。
- 从框架的角度来看，迁移是一次性的。由于成功的迁移会将配置以及通常的组件状态写入连接器，因此连接器在迁移完成后立即被视为已修改，后续迁移请求将被拒绝，直到连接器被丢弃并创建新的连接器。

=== 技术

ACAL Lynxus 的后端使用 Java 编写。Web 层使用 JAX-RS，并广泛使用 JavaScript 来提供用户界面。我们依赖多个第三方 JavaScript 库，包括 D3 和 Angular 等。我们使用 Apache Maven 进行构建，使用 Git 作为版本控制系统。

文档使用 link:http://asciidoctor.org[AsciiDoc^] 创建。