Deploy of new Data Processor on 0.99.0-BACKEND failed on port 8030

[leccher]

Hello to everybody.
I wrote a JsonFlattener Processor to flat complex JSON.

I have an apache streampipes backend in my docker compose with everything running fine but I noticed that A SP does not manage nested object in json

{
root:{
element:{
fields:[
{field1:"value1"},
{field2:"value2"}
]
}
}

So i decided "to flat" it.

{
root__element__fields__field1:"value1",
root__element__fields__field2:"value2",
}

and I wrote my JSON Flattener Processor for 0.99.0-SNAPSHOT A SP following the documentation I fund here:
https://streampipes.apache.org/docs/extend-tutorial-data-processors/

creating archetype using

mvn archetype:generate -DarchetypeGroupId=org.apache.streampipes \
-DarchetypeArtifactId=streampipes-archetype-extensions-jvm \
-DarchetypeVersion=0.99.0-SNAPSHOT \
-DartifactId=json-rebuilder \
-DclassNamePrefix=JsonRebuilder \
-DgroupId=org.streampipes.processor \
-DpackageName=rebuilder

Everything seems fine but when I run my extension container the processor cannot register itself into backend using 8030 backend servce:

json-flattener-1  | 2026-03-30T12:11:13.026Z  WARN   --- [           main] o.a.s.s.extensions.CoreRequestSubmitter  : Could not register service at core at url http://backend:8030 Trying again in 3 seconds

I noticed that the documentation is written for 0.98.0. Should I start from a data processor released form 0.99.0-SNAPSHOT I can find on github reposutory?

Thank you

[tenthe]

Hi @leccher,

thanks for reporting this.

Could you clarify how you are running your new processor, via Docker or directly from your IDE?

• If you are running it in Docker:
  Please make sure that the container of your extension is connected to the same Docker network as the StreamPipes backend. Otherwise, it won’t be able to reach the backend service at http://backend:8030.

• If you are running it from your IDE:
  You need to enable debug mode by setting the environment variable:
  SP_DEBUG=true
  (This should only be used for local development, as it reroutes traffic to localhost.)

  If the processor still tries to connect to http://backend:8030, you can additionally set:
  SP_HOST=<your-local-machine-ip>

This should allow your processor to correctly register with the backend.

Let me know if the issue persists

[tenthe]

Yes, one extension service can provide multiple data processors, and using one Init.java for both is the right approach. Your registration code looks correct in principle.

What should be checked is whether both processors use unique IDs in declareConfig(). If both processors use the same appId / element ID, one can overwrite the other and only one will appear in the UI.

Also, the “Extension Services” view shows the service itself, not necessarily each processor separately. The processors should appear in the extensions/installation view.

Could you share a bit more detail?

1. Is the service starting successfully?
2. Is one processor shown in the UI?
3. Does the service appear in the extension services view?
4. Are there any errors or warnings in the logs?

That should help narrow down the actual problem.

[leccher]

OK. Thank you.

Ok, Now I join them in one extensions and the answer is yess to all your questions.
In extension Services
Service ID: org.streampipes.processor.pe.jsonhelper-x54ya3
Service Group: org.streampipes.processor.pe.jsonhelper
Location: http://192.168.96.9:8090
Tags: custom transport:http
data_processor org.streampipes.processor.pe.jsonhelper.unflattener
data_processor org.streampipes.processor.pe.jsonhelper.flattener
sp_group org.streampipes.processor.pe.jsonhelper

In extensions I see both JSON Flattener and JSON Unflattener.

Here my repo
https://github.com/leccher/apache-streampipes-dataprocessor-json-helper.git

WHen I use it in my pipe I see a red icon

If I try to connect Adapter to JSON FLattener...

and in backend logs I see

backend-1     | 2026-04-15T12:01:51.071Z ERROR 1 --- [nio-8030-exec-5] o.a.s.rest.impl.PipelineResource         : Cannot invoke "org.apache.streampipes.model.grounding.EventGrounding.getTransportProtocols()" because "requirement" is null

I suppose something I missed in configurations....

    @Override
    public IDataProcessorConfiguration declareConfig() {
        StaticPropertyGroup group = new StaticPropertyGroup(
                "config",
                "JSON Flattener Config",
                "Configuration for JSON flattening",
                Arrays.asList(
                        new FreeTextStaticProperty(
                                SEP_KEY,
                                "Level separator",
                                "__"),
                        new FreeTextStaticProperty(
                                PREFIX_ARRAYS_KEY,
                                "Prefix for arrays",
                                "@@")));
        return DataProcessorConfiguration.create(
                JsonFlattenerDataProcessor::new,
                ProcessingElementBuilder.create("org.streampipes.processor.pe.jsonhelper.flattener", 0)
                        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
                        .withLocales(Locales.EN)
                        .category(DataProcessorType.TRANSFORM)
                        .requiredStream(StreamRequirementsBuilder
                                .create()
                                .requiredProperty(EpRequirements.anyProperty())
                                .build())
                        .requiredStaticProperty(group)
                        .outputStrategy(OutputStrategies.custom())
                        .build());
    }

    

    @Override
    public void onEvent(Event event, SpOutputCollector collector) {
        Map<String, Object> flat = new LinkedHashMap<>();
        flatten("", event.getRaw(), flat);

        Event output = EventFactory.fromMap(flat);
        collector.collect(output);
    }

I tried following documentation but I see it quite different from 0.99.0 (it is about 0.98)

Thank you.

[tenthe]

Thanks, this is likely an output strategy issue.

Please check the docs:
https://streampipes.apache.org/docs/extend-sdk-output-strategies/

For dynamic schemas, compare with FieldMapperProcessor:

• Uses OutputStrategies.customTransformation()
• Implements ResolvesContainerProvidedOutputStrategy
• Returns an EventSchema in resolveOutputStrategy(...) (see lines ~49 and ~75)

If your processor computes its schema, you should follow the same pattern instead of using a static strategy.

Regarding the static properties group: I’m not entirely sure what exactly you expect as input there, could you clarify that part?

[leccher]

Thank you tenthe.

I understood what, more or less I should do.

My Idea is to "flatten" every kind of json a pipe get in

In static properties group I ask for a string (a couple of chars) used to flatten the "json levels" in Event.raw (eg "__") and an identifier for array (used to identify them against maps in "rebuilding" process) eg @@.

📦 Example For Flattrener (conversely for Unflattener)

Input JSON

{
  "robot": {
    "info": { "speed": 3.4, "torque": 7.8 },
    "state": "busy",
    "speed_mode":["slow","normal","fast"]
  },
  "camera": {
    "quality": 0.92
  },
  "states":{
    "mode":["free","busy"]
  }
}

Flattened Output

{
  "robot__info__speed": 3.4,
  "robot__info__torque": 7.8,
  "robot__state": "busy",
  "robot__@@speed_mode__0": "slow",
  "robot__@@speed_mode__1": "normal",
  "robot__@@speed_mode__2": "fast",
  "camera__quality": 0.92,
  "states__@@mode__0":"free",
  "states__@@mode__1":"bosy"
}

As I understood in resolveOutputStrategy I should tell that new structure of event is now the flattened one.

Just look at ""Flattener"":

    @Override
    public void onEvent(Event event, SpOutputCollector collector) {
        Map<String, Object> flat = new LinkedHashMap<>();
        Map<String, Object> raw = event.getRaw();
        raw.put(EXTENSION_PREFIX + "-" + SEP_KEY, level_separator);
        raw.put(EXTENSION_PREFIX + "-" + PREFIX_ARRAYS_KEY, suffix_arrays);
        oldPropertyNames = raw.keySet().stream().toList();
        flatten("", raw, flat);
        currentEventStructure = new LinkedHashMap<>(flat);
        Event output = EventFactory.fromMap(flat);
        collector.collect(output);
    }

Following documentation you provided and FieldMapperProcessor code I wrote this:

    @Override
    public EventSchema resolveOutputStrategy(DataProcessorInvocation processingElement, ProcessingElementParameterExtractor extractor) {
        
        List<EventProperty> outputProps = new ArrayList<>();

        for (String key : currentEventStructure.keySet()) {
            outputProps.add(
                PrimitivePropertyBuilder
                    .create(inferDatatype(currentEventStructure.get(key)), key)
                    .build()
            );
        }

        return new EventSchema(outputProps);
    }

Here I just return the new EventSchema. Is it right?

Please, explain me more precisely what is needed.

Thank you.

[tenthe]

Hi @leccher,

the code in resolveOutputStrategy is called when the user configures the processing element. So you need to perform this function based on the input schema. You can get it like this:

EventSchema inputSchema = processingElement
    .getInputStreams()
    .get(0)
    .getEventSchema();

Based on this, you have to create the output schema with the builders, as you did in your example.

In your example, you used the variable currentEventStructure, which is set in the onEvent method. This will not work because onEvent is only called once the pipeline is started, so after the configuration.

Does this help?