Configuring sources using Spring Beans

    @Bean
    public DatastoreConfiguratorConsumer addFieldToStore() {
        return configurator -> {
            configurator.appendFields(SensiDatastoreDescriptionConfig.SCHEMA, StoreConstants.DYNAMIC_TENOR_STORE_NAME,
                    List.of(
                            new CustomField(DOUBLE_NUMBER_OF_DAYS, ILiteralType.DOUBLE),
                            new CustomField(TEST_FIELD, ILiteralType.STRING)
                    )
            );
        };
    }

@Qualifier(SP_QUALIFIER__SENSI_TOPIC_COLUMNS)
@Order(10)
public CsvColumnsProviderFunction<I> addFieldAndCalculator()

    @Bean
    @Qualifier(SP_QUALIFIER__SENSI_TOPIC_COLUMNS)
    @Order(10)
    public CsvColumnsProviderFunction<I> addFieldAndCalculator() {
        return (source, previousConfig) -> {
            // We retrieve the previous columns of the topic for the Tenors store.
            CsvTopicColumns prevColumns = previousConfig.get(StoreConstants.DYNAMIC_TENOR_STORE_NAME);
            List<String> newColumns = new ArrayList<>(prevColumns.getColumns());
            newColumns.add(TEST_FIELD);

            // We retrieve the previous calculators defined for the store.
            // While in this instance the topic name and the store are the same, calculators can be defined for multiple stores within a topic.
            Map<String, List<IColumnCalculator<ILineReader>>> topicCalculators = new HashMap<>(prevColumns.getCalculators());
            List<IColumnCalculator<ILineReader>> storeCalculators = new ArrayList<>(topicCalculators.get(StoreConstants.DYNAMIC_TENOR_STORE_NAME));
            storeCalculators.add(new IColumnCalculator<>() {
                @Override public String getColumnName() {
                    return DOUBLE_NUMBER_OF_DAYS;
                }

                @Override public Object compute(IColumnCalculationContext<ILineReader> context) {
                    return ((Double) context.getValue(StoreFieldNames.TENOR_NUMBER_OF_DAYS)) * 2.0;
                }
            });
            topicCalculators.put(StoreConstants.DYNAMIC_TENOR_STORE_NAME, storeCalculators);

            // We create a new CsvTopicColumns object with the new list of columns and the previously defined column calculators.
            return List.of(new CsvTopicColumns(
                    StoreConstants.DYNAMIC_TENOR_STORE_NAME,
                    newColumns,
                    topicCalculators
            ));
        };
    }

@Qualifier(SP_QUALIFIER__SENSI_TOPIC_PUBLISHERS)
    @Order(10)
    public CsvColumnsProviderFunction<I> addPublisher()

    @Bean
    @Qualifier(SP_QUALIFIER__SENSI_TOPIC_PUBLISHERS)
    @Order(10)
    public CsvColumnsProviderFunction<I> addPublisher(IDatastore datastore) {
        return (source, previousConfig) -> List.of(
                new CsvTopicPublisher<>(
                        StoreConstants.DYNAMIC_TENOR_STORE_NAME,
                        new ITuplePublisher<IFileInfo<Path>>() {
                            @Override
                            public void publish(IStoreMessage<? extends IFileInfo<Path>, ?> message, List<Object[]> tuples) {
                                // We use the expected index for the SensitivityName field here, while in a complete implementation the index would be an instance variable.
                                tuples.forEach(tuple -> tuple[1] = "TestValue");
                                datastore.getTransactionManager().addAll(StoreConstants.DYNAMIC_TENOR_STORE_NAME, tuples);
                            }

                            @Override
                            public Collection<String> getTargetStores() {
                                return Collections.singleton(StoreConstants.DYNAMIC_TENOR_STORE_NAME);
                            }
                        }
                )
        );
    }

@Configuration
public class CustomizationsConfig {
    private static final String TEST_FIELD = "TestData";
    private static final String DOUBLE_NUMBER_OF_DAYS = "DoubleNumberOfDays";

    @Bean
    public DatastoreConfiguratorConsumer addFieldToStore() {
        return configurator -> {
            configurator.appendFields(SensiDatastoreDescriptionConfig.SCHEMA, StoreConstants.DYNAMIC_TENOR_STORE_NAME,
                    List.of(
                            new CustomField(DOUBLE_NUMBER_OF_DAYS, ILiteralType.DOUBLE),
                            new CustomField(TEST_FIELD, ILiteralType.STRING)
                    )
            );
        };
    }

    @Bean
    @Qualifier(SP_QUALIFIER__SENSI_TOPIC_COLUMNS)
    @Order(10)
    public <I> CsvColumnsProviderFunction<I> addFieldAndCalculator() {
        return (source, previousConfig) -> {
            // We retrieve the previous columns of the topic for the Tenors store.
            CsvTopicColumns prevColumns = previousConfig.get(StoreNames.TENOR_STORE_NAME);
            List<String> newColumns = new ArrayList<>(prevColumns.getColumns());
            newColumns.add(TEST_FIELD);

            // We retrieve the previous calculators defined for the store. The Tenors store has a column calculator setting indices for each tenor.
            // While in this instance the topic name and the store are the same, calculators can be defined for multiple stores within a topic.
            Map<String, List<IColumnCalculator<ILineReader>>> topicCalculators = new HashMap<>(prevColumns.getCalculators());
            List<IColumnCalculator<ILineReader>> storeCalculators = new ArrayList<>(topicCalculators.get(StoreNames.TENOR_STORE_NAME));
            storeCalculators.add(new IColumnCalculator<>() {
                @Override public String getColumnName() {
                    return DOUBLE_NUMBER_OF_DAYS;
                }

                @Override public Object compute(IColumnCalculationContext<ILineReader> context) {
                    return ((Double) context.getValue(StoreFieldNames.TENOR_NUMBER_OF_DAYS)) * 2.0;
                }
            });
            topicCalculators.put(StoreNames.TENOR_STORE_NAME, storeCalculators);

            // We create a new CsvTopicColumns object with the new list of columns and the previously defined column calculators.
            return List.of(new CsvTopicColumns(
                    StoreNames.TENOR_STORE_NAME,
                    newColumns,
                    topicCalculators
            ));
        };
    }

    @Bean
    @Qualifier(SP_QUALIFIER__SENSI_TOPIC_PUBLISHERS)
    @Order(10)
    public <I> CsvPublisherProviderFunction<I> addPublisher(IDatastore datastore) {
        return (source, previousConfig) -> List.of(
                new CsvTopicPublisher<>(
                        StoreNames.TENOR_STORE_NAME,
                        new ITuplePublisher<IFileInfo<Path>>() {
                            @Override
                            public void publish(IStoreMessage<? extends IFileInfo<Path>, ?> message, List<Object[]> tuples) {
                                // We use the expected index for the SensitivityName field here, while in a complete implementation the index would be an instance variable.
                                tuples.forEach(tuple -> tuple[1] = "TestValue");
                                datastore.getTransactionManager().addAll(StoreNames.TENOR_STORE_NAME, tuples);
                            }

                            @Override
                            public Collection<String> getTargetStores() {
                                return Collections.singleton(StoreNames.TENOR_STORE_NAME);
                            }
                        }
                )
        );
    }
}

@Import(value = {
        ...
		CustomizationsConfig.class

})
public class MarketRiskConfig {
    ...
}

new ATableFormatTuplePublisher<>(datastore, Map.of(INSTRUMENT_ID, RISK_FACTOR)) {
    @Override
    public void publish(IStoreMessage<? extends IFileInfo<Path>, ?> message, List<Object[]> tuples) {
        Queue<Object[]> storeTuples = new ConcurrentLinkedQueue<>();
        tuples.forEach(
                tuple -> {
                    if (checkNull(tmmProps, message, tuple, TENOR_LABELS, TENOR_DATES, MATURITY_LABELS, MATURITY_DATES, MONEYNESS_LABELS,
                            NOMINAL)) {
                        Queue<Object[]> extractedScalarTuples = new ConcurrentLinkedQueue<>();
                        devectorizer.extractScalarTuples(message, tuple, extractedScalarTuples);
                        extractedScalarTuples.forEach(
                                scalarTuple -> storeTuples.add(createTupleForTable(message, SPOT_MARKET_DATA_STORE, scalarTuple))
                        );
                    }
                }
        );
        datastore.getTransactionManager().addAll(SPOT_MARKET_DATA_STORE, storeTuples);
    }

    @Override
    public Collection<String> getTargetStores() {
        return Set.of(SPOT_MARKET_DATA_STORE);
    }
}