Using field config in the DOFN

DoFn.java

 @ProcessElement
  public void processElement(@Element JsonNode jsonNode, ProcessContext context, PipelineOptions options) {

        FieldConfigurationOptions fieldConfigurationOptions = options.as(FieldConfigurationOptions.class);

        FieldConfigurationLoader fieldsConfigurationProvider = fieldConfigurationOptions.getFieldsConfigurationProvider();

        FieldsConfiguration config = fieldsConfigurationProvider.getConfig();
    
        TableRow tableRow = new TableRow();
    
        for (FieldConfig fieldConfig : config.getFields()) {
            setFieldByConfig(jsonNode, tableRow, fieldConfig);
        }
    
        context.output(tableRow);
 }

 private void setFieldByConfig(JsonNode jsonNode, TableRow tableRow, FieldConfig fieldConfig) {
   
     JsonNode childNode = getNodeByPath(fieldConfig.getJsonpath(), jsonNode);
     
     if( childNode != null ){
      
       switch (fieldConfig.getType()) {
           // converting the value by the type and set it in the tableRow by the fieldConfig.getColumnname()

       }
     }  
   
 }