Correct setting kafka graceful-shutdown property for dev and test profiles

[RoMiRoSSaN]

• Fixes Messaging kafka: incorrect setting of the graceful shutdown property for dev/test modes #48641

Fix for a minor bug that only exists in development or testing mode.
In some cases that are extremely difficult to reproduce, a situation occurs where channels are not formed correctly. For example, channel-in, channel-out are recognized as belonging to 1 specific input channel.
The problem is that 1 common list of Incoming, Outgoing and Channel annotations was formed, and the graceful-shutdown setting was set for them as for incoming channels.
According to the documentation only incoming channels have such a setting, but not for outgoing ones.
Therefore, you can remove the setting of the property for outgoing channels.

[ozangunalp]

Looks good!

Actually I need to check this, Channel annotations can be injected incoming channels too.

[quarkus-bot]

Status for workflow Quarkus CI

This is the status report for running Quarkus CI on commit e4002a8.

✅ The latest workflow run for the pull request has completed successfully.

It should be safe to merge provided you have a look at the other checks in the summary.

You can consult the Develocity build scans.

[gsmet]

@ozangunalp my understanding is that you decided it was all good, can we approve and merge? (just checking)

[ozangunalp]

No, please let me check the change locally and get back here asap

[ozangunalp]

Ok, looking closer at the code, there are multiple things:

While we apply graceful-shutdown=false for incoming channels, we can also set close-timeout=0 for outgoing ones, which is omitted.

Secondly, doing this only based on @Incoming channels would be incomplete; channels can be declared/injected using @Channel annotations, for both incoming and outgoing.

I think better way is to do it based on ConnectorManagedChannelBuildItem.

A not-so pseudo-code would look like :

    // visible for testing
    void disableGracefulShutdown(List<ConnectorManagedChannelBuildItem> channelsManagedByConnectors,
            BuildProducer<RunTimeConfigurationDefaultBuildItem> defaultConfigProducer,
            DefaultSerdeDiscoveryState discoveryState) {
        for (ConnectorManagedChannelBuildItem managed : channelsManagedByConnectors) {
            String channelName = managed.getName();
            boolean incoming = managed.getDirection() == INCOMING;
            if (!discoveryState.isKafkaConnector(channelsManagedByConnectors, incoming, channelName)) {
                continue;
            }
            String key = getChannelPropertyKey(channelName, incoming ? "graceful-shutdown" : "close-timeout", incoming);
            discoveryState.ifNotYetConfigured(key, () -> {
                defaultConfigProducer.produce(new RunTimeConfigurationDefaultBuildItem(key, incoming ? "false" : "0"));
            });
        }
    }

• Some unit tests wouldn't hurt.
  I should've looked at it sooner. Sorry for the long lead time.

@RoMiRoSSaN would this work for you?

[ozangunalp]

Proposed a cleaner way

[RoMiRoSSaN]

@ozangunalp Hi! Yes, I think it's not bad. It's just that in the current code, boolean incoming is always true, and my changes relied on that. I think your solution is definitely better.