anchor.diff:  }
   ----
   
  -To wrap the Read in a framework component you create a transform delegating to this one with a `@PartitionMapper` annotation
  -at least (you likely want to follow the best practices as well adding `@Icon` and `@Version`) and using `@Option` constructor injections
  -to configure the component:
  +To wrap the `Read` in a framework component, create a transform delegating to that Read with at least a `@PartitionMapper` annotation and using `@Option` constructor injections to configure the component. Also make sure to follow the best practices and to specify `@Icon` and `@Version`.
   
   [source,java]
   ----
  
positive.code: [[wrapping-a-beam-io__start]]
  == Limitations
  
  This part is limited to particular kinds of link:https://beam.apache.org/[Beam] `PTransform`:
  
  - the `PTransform<PBegin, PCollection<?>>` for the inputs
  - the `PTransform<PCollection<?>, PDone>` for the outputs. The outputs also must use a single (composite or not) `DoFn` in their `apply` method.
  
  == Wrap an input
  
  Assume you want to wrap an input like this one (based on existing Beam ones):
  
  [source,java]
  ----
  
negative.code: 
  == Advanced
  
  In terms of classloading, when you write an I/O, the Beam SDK Java core stack is assumed as provided in Talend Component Kit runtime. This way, you don't need to include it in the compile scope, it would be ignored anyway.
  
  === Coder
  
  If you need a JSonCoder, you can use the `org.talend.sdk.component.runtime.beam.factory.service.PluginCoderFactory` service,
  which gives you access to the JSON-P and JSON-B coders.
  
  There is also an Avro coder, which uses the `FileContainer`. It ensures it
  is self-contained for `IndexedRecord` and it does not require—as the default Apache Beam `AvroCoder`—to set the schema when creating a pipeline. +
  It consumes more space and therefore is slightly slower, but it is fine for `DoFn`, since it does not rely on serialization in most cases.
  See `org.talend.sdk.component.runtime.beam.transform.avro.IndexedRecordCoder`.
  
  === JsonObject to IndexedRecord
  
  The mainstream model is `JsonObject` but it is common to have a legacy system using
  `IndexedRecord`. To mitigate the transition, you can use the following `PTransforms`:
  
  - `IndexedRecordToJson`: to convert an `IndexedRecord` to a `JsonObject`.
  - `JsonToIndexedRecord`: to convert a `JsonObject` to an `IndexedRecord`.
  
  
  === Sample
  
  .Sample input based on Beam Kafka
  
  [source,java]
  ----
  