bug fixes: backend services patches - Implemented and deployed systemd-backed AMIs

2026-10-07 13:22:32 -04:00
parent 7135c81050
commit 019570d8a4
58 changed files with 1591 additions and 133 deletions
@@ -203,7 +203,7 @@ Run AWS-compatible services locally without an AWS account, auth token, or paid
 <details>
 <summary><strong>Real Docker where fidelity matters</strong></summary>
-Lambda, RDS, Neptune, ElastiCache, MSK, ECS, EC2, EKS, OpenSearch, CodeBuild, and Managed Service for Apache Flink use real Docker-backed execution instead of shallow mocks.
+Lambda, RDS, Neptune, ElastiCache, MSK, MSK Connect, ECS, EC2, EKS, OpenSearch, CodeBuild, and Managed Service for Apache Flink use real Docker-backed execution instead of shallow mocks.
 </details>
@@ -244,7 +244,7 @@ LocalStack's community edition [sunset in March 2026](https://blog.localstack.cl
 | License | MIT | Restricted |
 | API Gateway v2 / HTTP API | Yes | No |
 | Cognito | Yes | No |
-| RDS, ElastiCache, MSK | Real Docker | No |
+| RDS, ElastiCache, MSK, MSK Connect | Real Docker | No |
 | Neptune (graph DB + Gremlin WebSocket) | Real Docker | No |
 | DocumentDB (MongoDB-compatible) | Real Docker | No |
 | ECS, EC2, EKS | Real Docker | No |
@@ -298,7 +298,7 @@ Floci supports local emulation for application services, data services, eventing
 | Containers and compute | ECS, EC2, Lightsail, EKS, MWAA, ECR, EFS, CodeBuild, CodeDeploy, CodePipeline, CodeGuru Reviewer, CodeArtifact, AWS Batch, Auto Scaling, Application Auto Scaling, Elastic Beanstalk, ELB v2, ELB Classic |
 | Data, analytics, and AI | Athena, Glue, Lake Formation, EMR, EMR Serverless, Redshift, Redshift Data API, Firehose, Managed Service for Apache Flink, OpenSearch, S3 Tables, S3 Vectors, Textract, Transcribe, Comprehend, Rekognition, Translate, Bedrock, Bedrock Runtime, Bedrock AgentCore, Bedrock AgentCore Control, SageMaker |
 | Databases and caching | RDS, RDS Data API, Neptune, DocumentDB, DMS, MemoryDB, ElastiCache, Timestream for InfluxDB |
-| Messaging and transfer | SES, Kinesis, MSK, Amazon MQ, Transfer Family, DataSync, IoT Core, Amazon Connect, Amazon AppIntegrations |
+| Messaging and transfer | SES, Kinesis, MSK, MSK Connect, Amazon MQ, Transfer Family, DataSync, IoT Core, Amazon Connect, Amazon AppIntegrations |
 | Security and governance | AWS Network Firewall, AWS RAM, Service Quotas, WAF v2, GuardDuty, Amazon Inspector, CloudTrail, CloudFront, Resource Groups Tagging API, Resource Explorer 2, CloudHSM v2, Organizations, AWS Account Management, IAM Access Analyzer, IAM Identity Center (SSO Admin, OIDC, Access Portal, SCIM), Identity Store, Amazon Macie, Amazon Detective, Security Hub, Amazon Verified Permissions, Control Catalog, Control Tower, Service Catalog, AWS Marketplace |
 | Cost and billing | AWS Budgets, Pricing, Cost Explorer, Cost and Usage Reports, BCM Pricing Calculator, BCM Data Exports |
 | Resilience, backup, and config | AWS FIS, AWS Backup, AWS Config, AppConfig, AppConfigData, CloudFormation, Cloud Control API |
@@ -342,6 +342,7 @@ For operation-level compatibility, see the [Services Overview](https://floci.io/
 | Neptune | Real Docker | Graph DB via TinkerPop Gremlin Server (default) or Neo4j for openCypher/Bolt (`NEPTUNE_DB_TYPE`); RDS-shaped control plane; SigV4 proxy on port 8182 |
 | DocumentDB | Real Docker, mock mode available | MongoDB-compatible cluster via real MongoDB containers; RDS-shaped control plane; MongoDB wire protocol on port 27017 |
 | MSK | Real Docker | Kafka-compatible broker via Redpanda |
 | MSK Connect | Real Docker | Stock Kafka Connect workers with S3 custom plugin ZIPs |
 | Amazon MQ | Real Docker | RabbitMQ broker via rabbitmq:3-management; AMQP + management console |
 | Athena | In-process with DuckDB sidecar | Real SQL execution over S3 and Glue-backed views |
 | Glue | In-process | Data Catalog, Schema Registry, tables consumed by Athena |
@@ -0,0 +1,19 @@
 FROM docker.io/library/debian:trixie-slim
 RUN apt-get update \
    && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
        systemd \
        systemd-sysv \
        openssh-client \
        openssh-server \
        iproute2 \
        iptables \
        socat \
        curl \
        ca-certificates \
    && systemctl enable ssh \
    && rm -rf /var/lib/apt/lists/*
 ENV container=docker
 STOPSIGNAL SIGRTMIN+3
 CMD ["/sbin/init"]
@@ -0,0 +1,10 @@
 # Debian 13 EC2 guest
 Build the local guest image before launching `ami-debian13`:
 ```text
 docker build --platform linux/amd64 -t floci/ami-debian:13 docker/ec2/ami-images/debian-13
 ```
 The image runs systemd and starts SSH on every EC2 boot. Enable application services with
 `systemctl enable --now <service>` so they restart after `StopInstances` and `StartInstances`.
@@ -0,0 +1,17 @@
 FROM registry.fedoraproject.org/fedora:42
 RUN dnf install -y \
        systemd \
        openssh-clients \
        openssh-server \
        iproute \
        iptables \
        socat \
        curl \
        ca-certificates \
    && dnf clean all \
    && systemctl enable sshd
 ENV container=docker
 STOPSIGNAL SIGRTMIN+3
 CMD ["/sbin/init"]
@@ -0,0 +1,10 @@
 # Fedora 42 systemd EC2 guest
 Build the local guest image before launching `ami-fedora42`:
 ```text
 docker build --platform linux/amd64 -t floci/ami-fedora:42 docker/ec2/ami-images/fedora-42
 ```
 The image runs systemd and starts SSH on every EC2 boot. Enable application services with
 `systemctl enable --now <service>` so they restart after `StopInstances` and `StartInstances`.
@@ -0,0 +1,4 @@
 FROM registry.opensuse.org/opensuse/leap:15.6
 RUN zypper --non-interactive install -y openssh iproute2 iptables socat curl ca-certificates \
    && zypper clean --all
@@ -0,0 +1,9 @@
 # openSUSE Leap 15.6 EC2 guest
 Build the local guest image before launching `ami-opensuse-leap156`:
 ```text
 docker build --platform linux/amd64 -t floci/ami-opensuse:15.6 docker/ec2/ami-images/opensuse-leap-15.6
 ```
 The image includes OpenSSH and the EC2 metadata proxy dependencies.
@@ -0,0 +1,19 @@
 FROM public.ecr.aws/docker/library/ubuntu:20.04
 RUN apt-get update \
    && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
        systemd \
        systemd-sysv \
        openssh-client \
        openssh-server \
        iproute2 \
        iptables \
        socat \
        curl \
        ca-certificates \
    && systemctl enable ssh \
    && rm -rf /var/lib/apt/lists/*
 ENV container=docker
 STOPSIGNAL SIGRTMIN+3
 CMD ["/sbin/init"]
@@ -0,0 +1,10 @@
 # Ubuntu 20.04 systemd EC2 guest
 Build the local guest image before launching `ami-0abcdef1234567892`:
 ```text
 docker build --platform linux/amd64 -t floci/ami-ubuntu:20.04 docker/ec2/ami-images/ubuntu-20.04
 ```
 The image runs systemd and starts SSH on every EC2 boot. Enable application services with
 `systemctl enable --now <service>` so they restart after `StopInstances` and `StartInstances`.
@@ -26,14 +26,16 @@ metadata.
 | AMI ID | Aliases | Docker image |
 |---|---|---|
 | `ami-0abcdef1234567890` | `ami-amazonlinux2` | `public.ecr.aws/amazonlinux/amazonlinux:2` |
-| `ami-0abcdef1234567891` | `ami-amazonlinux2023` | `public.ecr.aws/amazonlinux/amazonlinux:2023` |
+| `ami-0abcdef1234567891` | `ami-amazonlinux2023` | `floci/ami-amazonlinux:2023` |
-| `ami-0abcdef1234567892` | `ami-ubuntu2004` | `public.ecr.aws/docker/library/ubuntu:20.04` |
+| `ami-0abcdef1234567892` | `ami-ubuntu2004` | `floci/ami-ubuntu:20.04` |
 | `ami-ubuntu2204` | | `public.ecr.aws/docker/library/ubuntu:22.04` |
 | `ami-ubuntu2404-arm64` | `ami-ubuntu2404` | `public.ecr.aws/docker/library/ubuntu:24.04` |
 | `ami-ubuntu2404-amd64` | | `public.ecr.aws/docker/library/ubuntu:24.04` |
 | `ami-ubuntu2404-cloud-arm64` | `ami-ubuntu2404-cloud` | `floci/ami-ubuntu:24.04-arm64` |
 | `ami-debian12` | | `public.ecr.aws/docker/library/debian:12` |
 | `ami-debian13` | | `floci/ami-debian:13` |
 | `ami-alpine` | | `public.ecr.aws/docker/library/alpine:latest` |
 | `ami-fedora42` | | `floci/ami-fedora:42` |
 | `ami-0abcdef1234567893` | | `public.ecr.aws/amazonlinux/amazonlinux:2023` |
 Any unrecognized AMI ID (including real AWS AMI IDs like `ami-0abc12345678`) falls back to the catalog `defaultDockerImage` (`public.ecr.aws/amazonlinux/amazonlinux:2023` by default).
@@ -65,6 +65,7 @@ Operation counts are exact. For dispatch-table services (Query and JSON 1.1) eac
 | [RDS Data API](rds-data.md) | `/Execute`, `/BeginTransaction`, `/CommitTransaction`, `/RollbackTransaction` | REST JSON | 4 |
 | [Timestream for InfluxDB](timestream-influxdb.md) | `POST /` + `X-Amz-Target: AmazonTimestreamInfluxDB.*` + InfluxDB container | JSON 1.0 + InfluxDB HTTP | 24 |
 | [MSK](msk.md) | `/v1/clusters/...`, `/api/v2/clusters/...` + Redpanda broker | REST JSON + Kafka | 8 |
 | [MSK Connect](kafka-connect.md) | `/v1/custom-plugins`, `/v1/connectors`, `/v1/worker-configurations` + Kafka Connect worker | REST JSON + Kafka Connect | 13 |
 | [Amazon MQ](amazonmq.md) | `/v1/brokers/...` + RabbitMQ broker | REST JSON + AMQP | 5 |
 | [MWAA](mwaa.md) | `/` REST paths for environments + CLI/web proxy | REST JSON | 10 |
 | [Athena](athena.md) | `POST /` + `X-Amz-Target: AmazonAthena.*` | JSON 1.1 | 4 |
@@ -0,0 +1,45 @@
 # MSK Connect
 **Protocol:** REST JSON
 **Signing name:** `kafkaconnect`
 Floci implements the MSK Connect control plane separately from MSK. Custom plugins are ZIP
 archives stored in Floci S3. A connector starts a pinned stock Apache Kafka Connect worker, extracts
 the referenced ZIP, and mounts only that extracted plugin directory into the worker.
 ## Supported Actions
 | Action | Description |
 |---|---|
 | `CreateCustomPlugin` | Register a ZIP custom plugin from an S3 bucket ARN and key |
 | `ListCustomPlugins` | List custom plugins |
 | `DescribeCustomPlugin` | Describe a custom plugin |
 | `DeleteCustomPlugin` | Delete a custom plugin |
 | `CreateWorkerConfiguration` | Create a worker properties configuration |
 | `ListWorkerConfigurations` | List worker configurations |
 | `DescribeWorkerConfiguration` | Describe a worker configuration |
 | `DeleteWorkerConfiguration` | Delete a worker configuration |
 | `CreateConnector` | Start a connector using a custom plugin |
 | `ListConnectors` | List connectors |
 | `DescribeConnector` | Describe a connector |
 | `UpdateConnector` | Update connector configuration |
 | `DeleteConnector` | Stop and delete a connector |
 ## Worker behavior
 Workers use the internal Redpanda address of a Floci-managed MSK cluster. The public bootstrap
 address supplied to `CreateConnector` is resolved to that cluster but is never passed into the
 container. This keeps the worker on Floci's Docker network while host clients continue using the
 public MSK bootstrap endpoint.
 No connector plugin is bundled. To run an S3 sink, upload a compatible S3 sink connector ZIP to
 Floci S3, create a custom plugin for it, and use that plugin in `CreateConnector`. The worker's
 container logs are streamed to CloudWatch Logs under `/aws/msk-connect/{connectorName}`.
 ## Configuration
 | Variable | Default | Description |
 |---|---|---|
 | `FLOCI_SERVICES_KAFKACONNECT_ENABLED` | `true` | Enable or disable MSK Connect |
 | `FLOCI_SERVICES_KAFKACONNECT_MOCK` | `false` | Persist control-plane metadata without starting workers |
 | `FLOCI_SERVICES_KAFKACONNECT_WORKER_IMAGE` | `apache/kafka:3.9.0` | Pinned stock Kafka Connect worker image |
@@ -91,6 +91,15 @@ TPS over 10 will trigger the LimitExceededException"), independent of the ten-ca
 limit above. Floci does not simulate real-time request throttling for this or any other
 action, so that limit is intentionally unenforced here.
 ## Current Limitations
 - `ListStreams`, `ListShards`, `ListTagsForStream`, and `ListStreamConsumers` do not yet
  support AWS pagination tokens. `ListShards` honors `MaxResults`, but does not return a
  continuation token.
 - Resource policy actions are not implemented.
 - KMS encryption settings are recorded in stream metadata, but record payloads are not
  cryptographically encrypted.
 ## Configuration
 | Variable | Default | Description |
@@ -142,6 +142,7 @@ nav:
    - RDS Data API: services/rds-data.md
    - Timestream for InfluxDB: services/timestream-influxdb.md
    - MSK (Kafka): services/msk.md
    - MSK Connect: services/kafka-connect.md
    - Amazon MQ: services/amazonmq.md
    - Glue: services/glue.md
    - Lake Formation: services/lakeformation.md
@@ -302,6 +302,7 @@ public interface EmulatorConfig {
    interface ServiceStorageOverrides {
        SsmStorageConfig ssm();
        KafkaConnectStorageConfig kafkaconnect();
        SqsStorageConfig sqs();
        S3StorageConfig s3();
        DynamoDbStorageConfig dynamodb();
@@ -365,6 +366,13 @@ public interface EmulatorConfig {
        long flushIntervalMs();
    }
    interface KafkaConnectStorageConfig {
        Optional<String> mode();
        @WithDefault("5000")
        long flushIntervalMs();
    }
    interface SqsStorageConfig {
        Optional<String> mode();
    }
@@ -673,6 +681,7 @@ public interface EmulatorConfig {
        ApiGatewayServiceConfig apigateway();
        IamServiceConfig iam();
        MskServiceConfig msk();
        KafkaConnectServiceConfig kafkaconnect();
        AmazonMqServiceConfig amazonmq();
        KinesisAnalyticsServiceConfig kinesisAnalytics();
        ElastiCacheServiceConfig elasticache();
@@ -1345,6 +1354,19 @@ public interface EmulatorConfig {
        int kafkaHostPortMax();
    }
    interface KafkaConnectServiceConfig {
        @WithDefault("true")
        boolean enabled();
        /** Metadata-only mode is useful where Docker is intentionally unavailable. */
        @WithDefault("false")
        boolean mock();
        /** Stock Apache Kafka distribution, never a connector-plugin image. */
        @WithDefault("apache/kafka:3.9.0")
        String workerImage();
    }
    interface AmazonMqServiceConfig {
        @WithDefault("true")
        boolean enabled();
@@ -146,6 +146,12 @@ public class ResolvedServiceCatalog {
                        "msk", config.storage().mode(), 5000L, null, ServiceProtocol.REST_JSON,
                        protocols(ServiceProtocol.REST_JSON),
                        Set.of(), Set.of("kafka"), Set.of(), Set.of(io.github.hectorvent.floci.services.msk.MskController.class)),
                descriptor("kafkaconnect", "kafkaconnect", config.services().kafkaconnect().enabled(), true,
                        "kafkaconnect", storageMode(config.storage().services().kafkaconnect().mode(), config.storage().mode()),
                        config.storage().services().kafkaconnect().flushIntervalMs(), null, ServiceProtocol.REST_JSON,
                        protocols(ServiceProtocol.REST_JSON),
                        Set.of(), Set.of("kafkaconnect"), Set.of(),
                        Set.of(io.github.hectorvent.floci.services.kafkaconnect.KafkaConnectController.class)),
                descriptor("mq", "amazonmq", config.services().amazonmq().enabled(), true,
                        "amazonmq", config.storage().mode(), 5000L, null, ServiceProtocol.REST_JSON,
                        protocols(ServiceProtocol.REST_JSON),
@@ -110,6 +110,10 @@ public class ContainerLifecycleManager {
        }
    }
    public String refreshImage(String image) {
        return imageCacheService.refreshImage(image);
    }
    /**
     * Creates a container without starting it. Use {@link #startCreated} to
     * start it after any pre-start setup (e.g. copying files into the
@@ -6,6 +6,7 @@ import io.github.hectorvent.floci.config.FlociCertificateAuthority;
 import io.github.hectorvent.floci.core.common.ContainerTeardown;
 import io.github.hectorvent.floci.core.common.ContainerTeardowns;
 import io.github.hectorvent.floci.core.common.ServiceRegistry;
 import io.github.hectorvent.floci.services.ec2.Ec2Service;
 import io.github.hectorvent.floci.lifecycle.inithook.InitializationHook;
 import jakarta.inject.Inject;
 import jakarta.ws.rs.GET;
@@ -44,15 +45,17 @@ public class EmulatorInfoController {
    private final Instance<ContainerTeardown> containerTeardowns;
    private final FlociCertificateAuthority certificateAuthority;
    private final EmulatorConfig config;
    private final Ec2Service ec2Service;
    @Inject
    public EmulatorInfoController(ServiceRegistry serviceRegistry,
                                  InitLifecycleState initLifecycleState,
                                  StorageFactory storageFactory,
                                  Instance<Resettable> resettables,
-                                  Instance<ContainerTeardown> containerTeardowns,
+                                   Instance<ContainerTeardown> containerTeardowns,
-                                  FlociCertificateAuthority certificateAuthority,
+                                   FlociCertificateAuthority certificateAuthority,
-                                  EmulatorConfig config) {
+                                   EmulatorConfig config,
                                   Ec2Service ec2Service) {
        this.serviceRegistry = serviceRegistry;
        this.initLifecycleState = initLifecycleState;
        this.storageFactory = storageFactory;
@@ -60,6 +63,7 @@ public class EmulatorInfoController {
        this.containerTeardowns = containerTeardowns;
        this.certificateAuthority = certificateAuthority;
        this.config = config;
        this.ec2Service = ec2Service;
        this.version = resolveVersion();
    }
@@ -104,7 +108,7 @@ public class EmulatorInfoController {
    @GET
    @Path("/diagnose")
    public Response diagnose() {
-        return Response.ok(Map.of()).build();
+        return Response.ok(Map.of("ec2", Map.of("instances", ec2Service.diagnosticInstances()))).build();
    }
    @GET
@@ -19,8 +19,11 @@ import io.github.hectorvent.floci.services.ecs.EcsService;
 import io.github.hectorvent.floci.services.ecs.model.CreateTaskSetRequest;
 import io.github.hectorvent.floci.services.ecs.model.EcsLoadBalancer;
 import io.github.hectorvent.floci.services.ecs.model.EcsServiceModel;
 import io.github.hectorvent.floci.services.ecs.model.LaunchType;
 import io.github.hectorvent.floci.services.ecs.model.TaskSet;
 import io.github.hectorvent.floci.services.elbv2.ElbV2Service;
 import io.github.hectorvent.floci.services.elbv2.model.Action;
 import io.github.hectorvent.floci.services.elbv2.model.Rule;
 import io.github.hectorvent.floci.services.elbv2.model.TargetGroup;
 import io.github.hectorvent.floci.services.elbv2.model.TargetHealth;
 import io.github.hectorvent.floci.services.lambda.LambdaService;
@@ -63,6 +66,8 @@ public class CodeDeployService {
    // Lambda/ECS validation hooks must call PutLifecycleEventHookExecutionStatus within one hour.
    private static final Duration DEFAULT_HOOK_CALLBACK_TIMEOUT = Duration.ofHours(1);
    private static final Duration STOP_POLL_INTERVAL = Duration.ofMillis(200);
    private static final int ECS_READINESS_MAX_POLLS = 150;
    private static final int ECS_TARGET_HEALTH_MAX_POLLS = 900;
    @FunctionalInterface
    interface Sleeper {
@@ -1543,6 +1548,12 @@ public class CodeDeployService {
                    createTaskSet.setLoadBalancers(List.of(loadBalancer));
                }
                greenTaskSet = ecsService.createTaskSet(createTaskSet, region);
                verifyGreenTaskSet(region, clusterName, serviceName, deploymentId, targetService, greenTaskSet,
                        greenTgArn);
                if (greenTaskSet.getLaunchType() == LaunchType.FARGATE) {
                    greenTaskSet = waitForGreenTaskSetRunning(region, clusterName, serviceName, greenTaskSet,
                            stopFlag);
                }
                appendTaskSetInfo(ecsTargetMap, greenTaskSet, greenTgArn, 0.0);
                finishLifecycleEvent(installEvent, "Succeeded");
            } catch (Exception e) {
@@ -1585,9 +1596,9 @@ public class CodeDeployService {
                if (stopFlag.get()) { finishEcsStopped(deployment, ecsTargetMap); return; }
                executeEcsAllowTraffic(region, deployment, configName, ecsTargetMap,
                        listenerArns, blueTgArn, greenTgArn, stopFlag);
                verifyListenersForwardToGreen(region, listenerArns, greenTgArn);
            } else {
-                Map<String, Object> allowEvent = addLifecycleEvent(ecsTargetMap, "AllowTraffic");
+                throw new IllegalStateException("ECS deployment group must have production listeners and blue/green target groups");
                finishLifecycleEvent(allowEvent, "Succeeded");
            }
            if (stopFlag.get()) { finishEcsStopped(deployment, ecsTargetMap); return; }
@@ -1615,10 +1626,11 @@ public class CodeDeployService {
                try {
                    ecsService.deleteTaskSet(clusterName, serviceName,
                            blueTaskSet.getTaskSetArn(), true, region);
                    verifyBlueTaskSetDrained(region, clusterName, serviceName, blueTaskSet.getTaskSetArn());
                    finishLifecycleEvent(terminateEvent, "Succeeded");
                } catch (Exception e) {
-                    LOG.debugv("Could not delete blue task set: {0}", e.getMessage());
+                    finishLifecycleEvent(terminateEvent, "Failed");
-                    finishLifecycleEvent(terminateEvent, "Succeeded");
+                    throw e;
                }
            }
@@ -1637,18 +1649,120 @@ public class CodeDeployService {
        }
    }
-    /** A real target group is switched only once every registered green task is routable. */
+    void verifyGreenTaskSet(String region, String clusterName, String serviceName, String deploymentId,
-    private void waitForGreenTargets(String region, String greenTgArn, AtomicBoolean stopFlag)
+                            EcsServiceModel targetService, TaskSet createdTaskSet, String greenTgArn) {
        if (createdTaskSet == null || createdTaskSet.getTaskSetArn() == null) {
            throw new IllegalStateException("ECS did not create a green task set");
        }
        List<TaskSet> taskSets = ecsService.describeTaskSets(clusterName, serviceName,
                List.of(createdTaskSet.getTaskSetArn()), region);
        if (taskSets.size() != 1) {
            throw new IllegalStateException("ECS did not retain the green task set");
        }
        TaskSet greenTaskSet = taskSets.getFirst();
        if (!deploymentId.equals(greenTaskSet.getExternalId())
                || !"CODE_DEPLOY".equals(greenTaskSet.getStartedBy())
                || !"ACTIVE".equals(greenTaskSet.getStatus())) {
            throw new IllegalStateException("ECS green task set is not a CodeDeploy active task set");
        }
        if (greenTaskSet.getLaunchType() != LaunchType.FARGATE && greenTaskSet.getLaunchType() != LaunchType.EC2) {
            throw new IllegalStateException("ECS green task set does not use FARGATE or EC2");
        }
        if (targetService.getLaunchType() != null && greenTaskSet.getLaunchType() != targetService.getLaunchType()) {
            throw new IllegalStateException("ECS green task set launch type does not match the service");
        }
        if (greenTaskSet.getLaunchType() == LaunchType.EC2
                && !hasRunningTasks(greenTaskSet)) {
            throw new IllegalStateException("ECS green task set has no running tasks");
        }
        if (greenTgArn == null || greenTaskSet.getLoadBalancers() == null || greenTaskSet.getLoadBalancers().stream()
                .noneMatch(loadBalancer -> greenTgArn.equals(loadBalancer.getTargetGroupArn()))) {
            throw new IllegalStateException("ECS green task set is not registered with the green target group");
        }
    }
    /** Fargate task launch is asynchronous, so the initial CreateTaskSet result is not yet stable. */
    TaskSet waitForGreenTaskSetRunning(String region, String clusterName, String serviceName,
                                       TaskSet createdTaskSet, AtomicBoolean stopFlag) throws InterruptedException {
        for (int poll = 0; poll < ECS_READINESS_MAX_POLLS && !stopFlag.get(); poll++) {
            List<TaskSet> taskSets = ecsService.describeTaskSets(clusterName, serviceName,
                    List.of(createdTaskSet.getTaskSetArn()), region);
            if (taskSets.size() == 1 && hasRunningTasks(taskSets.getFirst())) {
                return taskSets.getFirst();
            }
            sleeper.sleep(STOP_POLL_INTERVAL.toMillis());
        }
        if (!stopFlag.get()) {
            throw new IllegalStateException("ECS green task set has no running tasks");
        }
        return createdTaskSet;
    }
    private boolean hasRunningTasks(TaskSet taskSet) {
        return taskSet.getComputedDesiredCount() > 0
                && taskSet.getRunningCount() >= taskSet.getComputedDesiredCount();
    }
    /** A production listener is switched only once green tasks are registered and routable. */
    void waitForGreenTargets(String region, String greenTgArn, AtomicBoolean stopFlag)
            throws InterruptedException {
-        while (!stopFlag.get()) {
+        boolean registeredTargets = false;
        for (int poll = 0; poll < ECS_TARGET_HEALTH_MAX_POLLS && !stopFlag.get(); poll++) {
            List<TargetHealth> targets = elbV2Service.describeTargetHealth(region, greenTgArn, null);
-            // Mock ECS does not start containers or register target addresses. Preserve that control-plane
+            registeredTargets = !targets.isEmpty();
-            // behavior while Docker-backed deployments wait for the ELB health checker.
+            if (registeredTargets && targets.stream().allMatch(target -> "healthy".equals(target.getState()))) {
            if (targets.isEmpty() || targets.stream().allMatch(target -> "healthy".equals(target.getState()))) {
                return;
            }
            sleeper.sleep(STOP_POLL_INTERVAL.toMillis());
        }
        if (!stopFlag.get()) {
            throw new IllegalStateException(registeredTargets
                    ? "ECS green target group has unhealthy targets"
                    : "ECS green target group has no registered targets");
        }
    }
    void verifyListenersForwardToGreen(String region, List<String> listenerArns, String greenTgArn) {
        for (String listenerArn : listenerArns) {
            List<Rule> defaultRules = elbV2Service.describeRules(region, listenerArn, null).stream()
                    .filter(Rule::isDefault)
                    .toList();
            if (defaultRules.size() != 1 || !forwardsOnlyTo(defaultRules.getFirst().getActions(), greenTgArn)) {
                throw new IllegalStateException("Production listener did not switch to the green target group");
            }
        }
    }
    private boolean forwardsOnlyTo(List<Action> actions, String greenTgArn) {
        if (actions == null) {
            return false;
        }
        List<Action> forwardActions = actions.stream()
                .filter(action -> "forward".equals(action.getType()))
                .toList();
        return !forwardActions.isEmpty()
                && forwardActions.stream().allMatch(action -> forwardsOnlyTo(action, greenTgArn));
    }
    private boolean forwardsOnlyTo(Action action, String greenTgArn) {
        if (action.getTargetGroups() == null || action.getTargetGroups().isEmpty()) {
            return greenTgArn.equals(action.getTargetGroupArn());
        }
        boolean greenReceivesTraffic = action.getTargetGroups().stream()
                .anyMatch(target -> greenTgArn.equals(target.getTargetGroupArn()) && effectiveWeight(target) > 0);
        boolean otherTargetReceivesTraffic = action.getTargetGroups().stream()
                .anyMatch(target -> !greenTgArn.equals(target.getTargetGroupArn()) && effectiveWeight(target) > 0);
        return greenReceivesTraffic && !otherTargetReceivesTraffic;
    }
    private int effectiveWeight(Action.TargetGroupTuple target) {
        return target.getWeight() != null ? target.getWeight() : 1;
    }
    void verifyBlueTaskSetDrained(String region, String clusterName, String serviceName, String blueTaskSetArn) {
        if (!ecsService.describeTaskSets(clusterName, serviceName, List.of(blueTaskSetArn), region).isEmpty()) {
            throw new IllegalStateException("ECS blue task set was not drained");
        }
    }
    void executeEcsAllowTraffic(String region, Deployment deployment, String configName,
@@ -23,6 +23,7 @@ import io.github.hectorvent.floci.services.codepipeline.model.CodePipelinePipeli
 import io.github.hectorvent.floci.services.codepipeline.model.CodePipelinePipeline.TransitionState;
 import io.github.hectorvent.floci.services.codepipeline.model.CodePipelineStoredItem;
 import io.github.hectorvent.floci.services.ecs.EcsService;
 import io.github.hectorvent.floci.services.ecs.model.ContainerDefinition;
 import io.github.hectorvent.floci.services.ecs.model.RegisterTaskDefinitionRequest;
 import io.github.hectorvent.floci.services.ecs.model.TaskDefinition;
 import io.github.hectorvent.floci.services.lambda.LambdaService;
@@ -1325,10 +1326,10 @@ public class CodePipelineService {
        JsonNode config = action.path("configuration");
        String applicationName = requiredCodeDeployToEcsConfiguration(config, "ApplicationName");
        String deploymentGroupName = requiredCodeDeployToEcsConfiguration(config, "DeploymentGroupName");
-        String taskDefinitionPath = requiredCodeDeployToEcsConfiguration(config, "TaskDefinitionTemplatePath");
+        String taskDefinitionTemplate = readCodeDeployToEcsTemplate(execution, action, config,
-        String appSpecPath = requiredCodeDeployToEcsConfiguration(config, "AppSpecTemplatePath");
+                "TaskDefinitionTemplateArtifact", "TaskDefinitionTemplatePath");
-        String taskDefinitionTemplate = readCodeDeployToEcsTemplate(execution, action, taskDefinitionPath);
+        String appSpecTemplate = readCodeDeployToEcsTemplate(execution, action, config,
-        String appSpecTemplate = readCodeDeployToEcsTemplate(execution, action, appSpecPath);
+                "AppSpecTemplateArtifact", "AppSpecTemplatePath");
        RegisterTaskDefinitionRequest taskDefinitionRequest;
        try {
@@ -1341,6 +1342,7 @@ public class CodePipelineService {
            throw new AwsException("InvalidActionDeclarationException",
                    "Task definition template must specify family", 400);
        }
        substituteCodeDeployToEcsImage(execution, action, config, taskDefinitionRequest);
        TaskDefinition taskDefinition = ecsService.registerTaskDefinition(taskDefinitionRequest, execution.getRegion());
        String appSpec = appSpecTemplate.replace("<TASK_DEFINITION>", taskDefinition.getTaskDefinitionArn());
        Map<String, Object> revision = Map.of("revisionType", "AppSpecContent",
@@ -1351,13 +1353,25 @@ public class CodePipelineService {
        waitForCodeDeployDeployment(execution, state, deploymentId);
    }
-    private String readCodeDeployToEcsTemplate(CodePipelineExecution execution, JsonNode action, String reference) {
+    private String readCodeDeployToEcsTemplate(CodePipelineExecution execution, JsonNode action, JsonNode config,
-        int separator = reference.indexOf("::");
+                                               String artifactField, String pathField) {
-        if (separator <= 0 || separator == reference.length() - 2 || reference.indexOf("::", separator + 2) >= 0) {
+        String templatePath = requiredCodeDeployToEcsConfiguration(config, pathField);
-            throw new AwsException("InvalidActionDeclarationException",
+        int separator = templatePath.indexOf("::");
-                    "CodeDeployToECS template paths must use ArtifactName::TemplateFilePath", 400);
+        if (separator >= 0) {
            if (separator == 0 || separator == templatePath.length() - 2
                    || templatePath.indexOf("::", separator + 2) >= 0) {
                throw new AwsException("InvalidActionDeclarationException",
                        "CodeDeployToECS template paths must use ArtifactName::TemplateFilePath", 400);
            }
            return readCodeDeployToEcsTemplate(execution, action, templatePath.substring(0, separator),
                    templatePath.substring(separator + 2));
        }
-        String artifactName = reference.substring(0, separator);
+        String artifactName = requiredCodeDeployToEcsConfiguration(config, artifactField);
        return readCodeDeployToEcsTemplate(execution, action, artifactName, templatePath);
    }
    private String readCodeDeployToEcsTemplate(CodePipelineExecution execution, JsonNode action, String artifactName,
                                               String templatePath) {
        boolean declaredInput = false;
        for (JsonNode input : action.path("inputArtifacts")) {
            if (artifactName.equals(input.path("name").asText())) {
@@ -1373,7 +1387,7 @@ public class CodePipelineService {
        if (artifact == null) {
            throw new AwsException("InvalidJobStateException", "Input artifact is not available: " + artifactName, 400);
        }
-        return readTemplateFromArtifact(artifact, reference.substring(separator + 2));
+        return readTemplateFromArtifact(artifact, templatePath);
    }
    private static String requiredCodeDeployToEcsConfiguration(JsonNode config, String field) {
@@ -1385,6 +1399,40 @@ public class CodePipelineService {
        return value;
    }
    private void substituteCodeDeployToEcsImage(CodePipelineExecution execution, JsonNode action, JsonNode config,
                                                RegisterTaskDefinitionRequest taskDefinitionRequest) {
        String imageArtifactName = requiredCodeDeployToEcsConfiguration(config, "Image1ArtifactName");
        String containerName = requiredCodeDeployToEcsConfiguration(config, "Image1ContainerName");
        String imageDetail = readCodeDeployToEcsTemplate(execution, action, imageArtifactName, "imageDetail.json");
        JsonNode imageDetailJson;
        try {
            imageDetailJson = mapper.readTree(imageDetail);
        } catch (java.io.IOException exception) {
            throw new AwsException("InvalidActionDeclarationException",
                    "imageDetail.json must contain a valid image detail JSON object", 400);
        }
        String imageUri = imageDetailJson.path("ImageURI").asText(null);
        if (imageUri == null || imageUri.isBlank()) {
            throw new AwsException("InvalidActionDeclarationException", "imageDetail.json must specify ImageURI", 400);
        }
        if (taskDefinitionRequest.getContainerDefinitions() == null) {
            throw new AwsException("InvalidActionDeclarationException",
                    "Task definition template must specify containerDefinitions", 400);
        }
        for (ContainerDefinition container : taskDefinitionRequest.getContainerDefinitions()) {
            if (containerName.equals(container.getName())) {
                if (container.getImage() == null) {
                    throw new AwsException("InvalidActionDeclarationException",
                            "Image1ContainerName must identify a container with an image", 400);
                }
                container.setImage(container.getImage().replace("<IMAGE1_NAME>", imageUri));
                return;
            }
        }
        throw new AwsException("InvalidActionDeclarationException",
                "Image1ContainerName does not match a task definition container", 400);
    }
    private void waitForCodeDeployDeployment(CodePipelineExecution execution, ActionExecution state, String deploymentId)
            throws InterruptedException {
        Deployment deployment = codeDeployService.getDeployment(execution.getRegion(), deploymentId);
@@ -510,12 +510,11 @@ public class Ec2ContainerManager {
                if (publicKey != null && !publicKey.isBlank()) {
                    injectSshKey(containerId, publicKey);
                }
-                // sshd runs on every instance regardless of whether a key pair was supplied,
+                // Systemd-backed AMIs start their enabled SSH unit during boot. Starting a second
-                // matching real AWS AMIs (which start it as part of normal boot, independent of
+                // daemon here races it for port 22; lightweight guests still need this bootstrap.
-                // key-pair association) - starting it only when a key was present meant
+                if (!image.systemd()) {
-                // run-instances without --key-name produced a "connection refused" instead of AWS's
+                    startSshd(containerId, instanceId);
-                // actual behavior: a running daemon with simply nothing to authenticate against.
+                }
                startSshd(containerId, instanceId);
                // Execute UserData
                String userData = instance.getUserData();
@@ -1793,6 +1792,7 @@ public class Ec2ContainerManager {
                    // and sshd is never installed.
                    "  elif command -v yum >/dev/null 2>&1; then yum install -y openssh-server openssh-clients >/dev/null 2>&1;" +
                    "  elif command -v apt-get >/dev/null 2>&1; then DEBIAN_FRONTEND=noninteractive apt-get install -y openssh-server openssh-client >/dev/null 2>&1;" +
                    "  elif command -v zypper >/dev/null 2>&1; then zypper --non-interactive install -y openssh >/dev/null 2>&1;" +
                    "  elif command -v apk >/dev/null 2>&1; then apk add --no-cache openssh >/dev/null 2>&1;" +
                    "  fi;" +
                    "fi;" +
@@ -35,6 +35,8 @@ public final class Ec2MetadataProxy {
                // leaving the instance without a link-local IMDS endpoint.
                "elif command -v yum >/dev/null 2>&1; then",
                "  yum install -y iproute iptables socat curl ca-certificates >/dev/null",
                "elif command -v zypper >/dev/null 2>&1; then",
                "  zypper --non-interactive install -y iproute2 iptables socat curl ca-certificates >/dev/null",
                "elif command -v apk >/dev/null 2>&1; then",
                "  apk add --no-cache iproute2 iptables socat curl ca-certificates >/dev/null",
                "else",
@@ -702,6 +702,15 @@ public class Ec2QueryHandler {
        int networkInterfaceDeviceIndex = parseIntParam(p, "NetworkInterface.1.DeviceIndex", 0);
        String clientToken = p.getFirst("ClientToken");
        List<String> sgIds = getList(p, "SecurityGroupId");
        if (sgIds.isEmpty()) {
            sgIds = getList(p, "NetworkInterface.1.SecurityGroupId");
        }
        if (sgIds.isEmpty()) {
            sgIds = getList(p, "NetworkInterface.1.Groups");
        }
        if (sgIds.isEmpty()) {
            sgIds = getList(p, "NetworkInterface.1.GroupId");
        }
        // UserData is base64-encoded in the wire format
        String userDataEncoded = p.getFirst("UserData");
@@ -5005,6 +5005,21 @@ public class Ec2Service implements ContainerTeardown, ResourceProvider {
        return containerManager.isContainerRunning(instance.getDockerContainerId());
    }
    public List<Map<String, Object>> diagnosticInstances() {
        return instances.scan(key -> true).stream().map(instance -> {
            boolean containerRunning = isInstanceContainerRunning(instance.getInstanceId());
            String containerStatus = instance.getDockerContainerId() == null
                    ? "unavailable" : containerRunning ? "running" : "stopped";
            Map<String, Object> diagnostic = new LinkedHashMap<>();
            diagnostic.put("instanceId", instance.getInstanceId());
            diagnostic.put("sshHost", "localhost");
            diagnostic.put("sshPort", instance.getSshHostPort());
            diagnostic.put("containerId", instance.getDockerContainerId());
            diagnostic.put("containerStatus", containerStatus);
            return diagnostic;
        }).toList();
    }
    private static void requireKeyName(String keyName) {
        if (keyName == null || keyName.isBlank()) {
            throw new AwsException("MissingParameter", "The request must contain the parameter KeyName", 400);
@@ -226,6 +226,9 @@ public class EcsContainerManager {
                    metadataId));
            imagesByContainer.put(def, ecrRegistryManager.rewriteImageUri(def.getImage()));
        }
        for (String image : new HashSet<>(imagesByContainer.values())) {
            lifecycleManager.refreshImage(image);
        }
        PreparedNetwork protectedNetwork = prepareNetwork(task, taskDef, region, taskId);
@@ -1099,6 +1099,7 @@ public class ElbV2Service implements ResourceProvider {
        if (defaultRule == null) {
            return;
        }
        Listener listener = requireListener(region, listenerArn);
        Action action = new Action();
        action.setType("forward");
        if (greenWeightPct >= 100) {
@@ -1113,6 +1114,9 @@ public class ElbV2Service implements ResourceProvider {
            action.setTargetGroups(List.of(blueTuple, greenTuple));
        }
        defaultRule.setActions(List.of(action));
        listener.setDefaultActions(List.of(action));
        persistRegion(rules, region);
        persistRegion(listeners, region);
        dataPlane.recompileRules(listenerArn, getListenerRules(region, listenerArn));
    }
@@ -0,0 +1,73 @@
 package io.github.hectorvent.floci.services.kafkaconnect;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectConnector;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectCustomPlugin;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectWorkerConfiguration;
 import jakarta.inject.Inject;
 import jakarta.ws.rs.Consumes;
 import jakarta.ws.rs.DELETE;
 import jakarta.ws.rs.GET;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.PUT;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.PathParam;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 /** AWS MSK Connect REST JSON control plane. */
@Path("/")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
 public class KafkaConnectController {
    private final KafkaConnectService service;
    private final ObjectMapper objectMapper;
    @Inject
    public KafkaConnectController(KafkaConnectService service, ObjectMapper objectMapper) { this.service = service; this.objectMapper = objectMapper; }
    @POST @Path("/v1/custom-plugins")
    public Response createCustomPlugin(JsonNode request) {
        JsonNode location = request.path("location").path("s3Location");
        KafkaConnectCustomPlugin plugin = service.createPlugin(text(request, "name"), text(request, "contentType"),
                text(location, "bucketArn"), text(location, "fileKey"), strings(request.get("tags")));
        return Response.ok(pluginView(plugin)).build();
    }
    @GET @Path("/v1/custom-plugins") public Response listCustomPlugins() { return Response.ok(Map.of("customPlugins", service.plugins().stream().map(this::pluginView).toList())).build(); }
    @GET @Path("/v1/custom-plugins/{arn}") public Response describeCustomPlugin(@PathParam("arn") String arn) { return Response.ok(pluginView(service.plugin(arn))).build(); }
    @DELETE @Path("/v1/custom-plugins/{arn}") public Response deleteCustomPlugin(@PathParam("arn") String arn) { service.deletePlugin(arn); return Response.ok(Map.of("customPluginArn", arn)).build(); }
    @POST @Path("/v1/worker-configurations") public Response createWorkerConfiguration(JsonNode request) { return Response.ok(workerView(service.createWorkerConfiguration(text(request, "name"), text(request, "propertiesFileContent")))).build(); }
    @GET @Path("/v1/worker-configurations") public Response listWorkerConfigurations() { return Response.ok(Map.of("workerConfigurations", service.workers().stream().map(this::workerView).toList())).build(); }
    @GET @Path("/v1/worker-configurations/{arn}") public Response describeWorkerConfiguration(@PathParam("arn") String arn) { return Response.ok(workerView(service.worker(arn))).build(); }
    @DELETE @Path("/v1/worker-configurations/{arn}") public Response deleteWorkerConfiguration(@PathParam("arn") String arn) { service.deleteWorker(arn); return Response.ok(Map.of("workerConfigurationArn", arn)).build(); }
    @POST @Path("/v1/connectors")
    public Response createConnector(JsonNode request) {
        JsonNode cluster = request.path("kafkaCluster").path("apacheKafkaCluster");
        JsonNode plugin = request.path("plugins");
        String pluginArn = plugin.isArray() && !plugin.isEmpty() ? text(plugin.get(0).path("customPlugin"), "customPluginArn") : null;
        JsonNode worker = request.path("workerConfiguration");
        String bootstrapServers = cluster.path("bootstrapServers").isArray()
                ? cluster.path("bootstrapServers").path(0).asText(null) : text(cluster, "bootstrapServers");
        KafkaConnectConnector connector = service.createConnector(text(request, "connectorName"), bootstrapServers, pluginArn,
                text(worker, "workerConfigurationArn"), strings(request.get("connectorConfiguration")), strings(request.get("tags")));
        return Response.ok(connectorView(connector)).build();
    }
    @GET @Path("/v1/connectors") public Response listConnectors() { return Response.ok(Map.of("connectors", service.connectors().stream().map(this::connectorView).toList())).build(); }
    @GET @Path("/v1/connectors/{arn}") public Response describeConnector(@PathParam("arn") String arn) { return Response.ok(connectorView(service.connector(arn))).build(); }
    @PUT @Path("/v1/connectors/{arn}") public Response updateConnector(@PathParam("arn") String arn, JsonNode request) { return Response.ok(connectorView(service.updateConnector(arn, strings(request.get("connectorConfiguration"))))).build(); }
    @DELETE @Path("/v1/connectors/{arn}") public Response deleteConnector(@PathParam("arn") String arn) { service.deleteConnector(arn); return Response.ok(Map.of("connectorArn", arn)).build(); }
    private Map<String, Object> pluginView(KafkaConnectCustomPlugin plugin) { return Map.of("customPluginArn", plugin.getArn(), "name", plugin.getName(), "customPluginName", plugin.getName(), "contentType", plugin.getContentType(), "creationTime", plugin.getCreationTime(), "state", "ACTIVE", "contentLocation", Map.of("s3Location", Map.of("bucketArn", plugin.getBucketArn(), "fileKey", plugin.getFileKey()))); }
    private Map<String, Object> workerView(KafkaConnectWorkerConfiguration worker) { return Map.of("workerConfigurationArn", worker.getArn(), "name", worker.getName(), "revision", worker.getRevision(), "creationTime", worker.getCreationTime(), "propertiesFileContent", worker.getPropertiesFileContent()); }
    private Map<String, Object> connectorView(KafkaConnectConnector connector) { Map<String, Object> view = new LinkedHashMap<>(); view.put("connectorArn", connector.getArn()); view.put("connectorName", connector.getName()); view.put("connectorState", connector.getState()); view.put("currentVersion", String.valueOf(connector.getCurrentVersion())); view.put("creationTime", connector.getCreationTime()); view.put("connectorConfiguration", connector.getConnectorConfiguration()); return view; }
    private String text(JsonNode node, String name) { JsonNode value = node.get(name); return value == null || value.isNull() ? null : value.asText(); }
    private Map<String, String> strings(JsonNode node) { Map<String, String> values = new LinkedHashMap<>(); if (node != null && node.isObject()) node.fields().forEachRemaining(entry -> values.put(entry.getKey(), entry.getValue().asText())); return values; }
 }
@@ -0,0 +1,125 @@
 package io.github.hectorvent.floci.services.kafkaconnect;
 import com.fasterxml.jackson.core.type.TypeReference;
 import io.github.hectorvent.floci.config.EmulatorConfig;
 import io.github.hectorvent.floci.core.common.AwsArnUtils;
 import io.github.hectorvent.floci.core.common.AwsException;
 import io.github.hectorvent.floci.core.common.RegionResolver;
 import io.github.hectorvent.floci.core.storage.StorageBackend;
 import io.github.hectorvent.floci.core.storage.StorageFactory;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectConnector;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectCustomPlugin;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectWorkerConfiguration;
 import io.github.hectorvent.floci.services.msk.MskService;
 import io.github.hectorvent.floci.services.msk.model.MskCluster;
 import io.github.hectorvent.floci.services.s3.S3Service;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.jboss.logging.Logger;
 import java.time.Instant;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@ApplicationScoped
 public class KafkaConnectService {
    private static final Logger LOG = Logger.getLogger(KafkaConnectService.class);
    private final StorageBackend<String, KafkaConnectCustomPlugin> plugins;
    private final StorageBackend<String, KafkaConnectConnector> connectors;
    private final StorageBackend<String, KafkaConnectWorkerConfiguration> workers;
    private final RegionResolver regionResolver;
    private final S3Service s3Service;
    private final MskService mskService;
    private final KafkaConnectWorkerManager workerManager;
    private final EmulatorConfig config;
    @Inject
    public KafkaConnectService(StorageFactory storageFactory, RegionResolver regionResolver, S3Service s3Service,
                               MskService mskService, KafkaConnectWorkerManager workerManager, EmulatorConfig config) {
        this.plugins = storageFactory.create("kafkaconnect", "custom-plugins.json", new TypeReference<Map<String, KafkaConnectCustomPlugin>>() { });
        this.connectors = storageFactory.create("kafkaconnect", "connectors.json", new TypeReference<Map<String, KafkaConnectConnector>>() { });
        this.workers = storageFactory.create("kafkaconnect", "worker-configurations.json", new TypeReference<Map<String, KafkaConnectWorkerConfiguration>>() { });
        this.regionResolver = regionResolver;
        this.s3Service = s3Service;
        this.mskService = mskService;
        this.workerManager = workerManager;
        this.config = config;
    }
    public KafkaConnectCustomPlugin createPlugin(String name, String contentType, String bucketArn, String fileKey, Map<String, String> tags) {
        require(name, "customPluginName"); require(bucketArn, "contentLocation.s3Location.bucketArn"); require(fileKey, "contentLocation.s3Location.fileKey");
        if (!"JAR".equals(contentType)) { throw badRequest("contentType must be JAR"); }
        String bucket = bucketName(bucketArn);
        s3Service.getObject(bucket, fileKey);
        KafkaConnectCustomPlugin plugin = new KafkaConnectCustomPlugin();
        plugin.setArn(arn("custom-plugin", name)); plugin.setName(name); plugin.setContentType(contentType);
        plugin.setBucketArn(bucketArn); plugin.setFileKey(fileKey); plugin.setCreationTime(Instant.now()); plugin.setTags(tags);
        plugins.put(plugin.getArn(), plugin);
        return plugin;
    }
    public KafkaConnectWorkerConfiguration createWorkerConfiguration(String name, String properties) {
        require(name, "name"); require(properties, "propertiesFileContent");
        KafkaConnectWorkerConfiguration worker = new KafkaConnectWorkerConfiguration();
        worker.setArn(arn("worker-configuration", name)); worker.setName(name); worker.setPropertiesFileContent(properties);
        worker.setRevision(1); worker.setCreationTime(Instant.now()); workers.put(worker.getArn(), worker);
        return worker;
    }
    public KafkaConnectConnector createConnector(String name, String bootstrapServers, String pluginArn, String workerArn,
                                                  Map<String, String> connectorConfiguration, Map<String, String> tags) {
        require(name, "connectorName"); require(bootstrapServers, "kafkaCluster.apacheKafkaCluster.bootstrapServers"); require(pluginArn, "plugins");
        KafkaConnectCustomPlugin plugin = plugin(pluginArn);
        MskCluster cluster = mskService.listClusters().stream()
                .filter(candidate -> bootstrapServers.equals(candidate.getBootstrapBrokers()))
                .findFirst().orElseThrow(() -> badRequest("bootstrapServers must name an MSK cluster managed by Floci"));
        if (cluster.getInternalBootstrapBrokers() == null && !config.services().kafkaconnect().mock()) {
            throw badRequest("The MSK cluster does not have an internal Redpanda bootstrap endpoint");
        }
        KafkaConnectConnector connector = new KafkaConnectConnector();
        applyS3Defaults(connectorConfiguration);
        connector.setArn(arn("connector", name)); connector.setName(name); connector.setState("RUNNING");
        connector.setClusterArn(cluster.getClusterArn()); connector.setPluginArn(pluginArn); connector.setWorkerConfigurationArn(workerArn);
        connector.setCurrentVersion(1); connector.setCreationTime(Instant.now()); connector.setConnectorConfiguration(connectorConfiguration); connector.setTags(tags);
        if (!config.services().kafkaconnect().mock()) {
            String properties = workerArn == null ? null : worker(workerArn).getPropertiesFileContent();
            byte[] archive = s3Service.getObject(bucketName(plugin.getBucketArn()), plugin.getFileKey()).getData();
            workerManager.start(connector, plugin, archive, cluster.getInternalBootstrapBrokers(), properties,
                    regionResolver.getRegion(), regionResolver.getAccountId());
        }
        connectors.put(connector.getArn(), connector);
        return connector;
    }
    public KafkaConnectConnector updateConnector(String arn, Map<String, String> configuration) {
        KafkaConnectConnector connector = connector(arn);
        Map<String, String> updatedConfiguration = new LinkedHashMap<>(connector.getConnectorConfiguration());
        updatedConfiguration.putAll(configuration);
        applyS3Defaults(updatedConfiguration);
        connector.setConnectorConfiguration(updatedConfiguration); connector.setCurrentVersion(connector.getCurrentVersion() + 1);
        connectors.put(arn, connector); if (!config.services().kafkaconnect().mock()) { workerManager.configure(connector); } return connector;
    }
    public void deleteConnector(String arn) { connector(arn); workerManager.stop(arn); connectors.delete(arn); }
    public void deletePlugin(String arn) { plugin(arn); plugins.delete(arn); }
    public void deleteWorker(String arn) { worker(arn); workers.delete(arn); }
    public KafkaConnectCustomPlugin plugin(String arn) { return plugins.get(arn).orElseThrow(() -> notFound("Custom plugin", arn)); }
    public KafkaConnectConnector connector(String arn) { return connectors.get(arn).orElseThrow(() -> notFound("Connector", arn)); }
    public KafkaConnectWorkerConfiguration worker(String arn) { return workers.get(arn).orElseThrow(() -> notFound("Worker configuration", arn)); }
    public List<KafkaConnectCustomPlugin> plugins() { return plugins.scan(key -> true); }
    public List<KafkaConnectConnector> connectors() { return connectors.scan(key -> true); }
    public List<KafkaConnectWorkerConfiguration> workers() { return workers.scan(key -> true); }
    private void applyS3Defaults(Map<String, String> configuration) {
        configuration.putIfAbsent("store.url", config.baseUrl());
        configuration.putIfAbsent("aws.access.key.id", "test");
        configuration.putIfAbsent("aws.secret.access.key", "test");
        configuration.putIfAbsent("format.class", "io.confluent.connect.s3.format.json.JsonFormat");
        configuration.putIfAbsent("storage.class", "io.confluent.connect.s3.storage.S3Storage");
    }
    private String arn(String type, String name) { return regionResolver.buildArn("kafkaconnect", regionResolver.getRegion(), type + "/" + name + "/" + UUID.randomUUID()); }
    private String bucketName(String bucketArn) { try { String resource = AwsArnUtils.parse(bucketArn).resource(); if (!"s3".equals(AwsArnUtils.parse(bucketArn).service()) || resource.isBlank()) throw new IllegalArgumentException(); return resource; } catch (IllegalArgumentException e) { throw badRequest("contentLocation.s3Location.bucketArn must be an S3 bucket ARN"); } }
    private void require(String value, String field) { if (value == null || value.isBlank()) throw badRequest(field + " is required"); }
    private AwsException badRequest(String message) { return new AwsException("BadRequestException", message, 400); }
    private AwsException notFound(String type, String arn) { return new AwsException("NotFoundException", type + " " + arn + " does not exist.", 404); }
 }
@@ -0,0 +1,197 @@
 package io.github.hectorvent.floci.services.kafkaconnect;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.github.hectorvent.floci.config.EmulatorConfig;
 import io.github.hectorvent.floci.core.common.docker.ContainerBuilder;
 import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager;
 import io.github.hectorvent.floci.core.common.docker.ContainerLogStreamer;
 import io.github.hectorvent.floci.core.common.docker.ContainerSpec;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectConnector;
 import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectCustomPlugin;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.jboss.logging.Logger;
 import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
 import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.Closeable;
 import java.io.IOException;
 import java.net.HttpURLConnection;
 import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.zip.ZipInputStream;
 /** Runs stock Kafka Connect workers. Connector implementations are supplied exclusively by S3 ZIPs. */
@ApplicationScoped
 public class KafkaConnectWorkerManager {
    private static final Logger LOG = Logger.getLogger(KafkaConnectWorkerManager.class);
    private static final int REST_PORT = 8083;
    private final ContainerBuilder containerBuilder;
    private final ContainerLifecycleManager lifecycleManager;
    private final ContainerLogStreamer logStreamer;
    private final EmulatorConfig config;
    private final ObjectMapper objectMapper;
    private final Map<String, Worker> workers = new ConcurrentHashMap<>();
    @Inject
    public KafkaConnectWorkerManager(ContainerBuilder containerBuilder, ContainerLifecycleManager lifecycleManager,
                                     ContainerLogStreamer logStreamer, EmulatorConfig config, ObjectMapper objectMapper) {
        this.containerBuilder = containerBuilder;
        this.lifecycleManager = lifecycleManager;
        this.logStreamer = logStreamer;
        this.config = config;
        this.objectMapper = objectMapper;
    }
    public void start(KafkaConnectConnector connector, KafkaConnectCustomPlugin plugin, byte[] archive,
                      String internalBootstrap, String workerProperties, String region, String account) {
        try {
            Path directory = Files.createTempDirectory("floci-kafka-connect-");
            Path plugins = Files.createDirectories(directory.resolve("plugins"));
            extractZip(archive, plugins);
            Path properties = directory.resolve("connect-distributed.properties");
            Files.writeString(properties, distributedProperties(internalBootstrap, workerProperties));
            String name = containerName(connector.getArn());
            lifecycleManager.removeIfExists(name);
            ContainerSpec spec = containerBuilder.newContainer(config.services().kafkaconnect().workerImage())
                    .withName(name).withDockerNetwork(config.services().dockerNetwork()).withLogRotation()
                    .withDynamicPort(REST_PORT)
                    .withCmd(List.of("/opt/kafka/bin/connect-distributed.sh", "/opt/floci/connect-distributed.properties"))
                    .build();
            String containerId = lifecycleManager.create(spec);
            copyWorkDirectory(containerId, directory);
            ContainerLifecycleManager.ContainerInfo info = lifecycleManager.startCreated(containerId, spec);
            String endpoint = "http://" + info.getEndpoint(REST_PORT).host() + ":" + info.getEndpoint(REST_PORT).port();
            Closeable logs = logStreamer.attachForAccount(account, info.containerId(), "/aws/msk-connect/" + connector.getName(),
                    logStreamer.generateLogStreamName(info.containerId().substring(0, 8)), region, "kafkaconnect:" + connector.getName());
            workers.put(connector.getArn(), new Worker(info.containerId(), endpoint, directory, logs));
            try {
                configure(connector);
            } catch (RuntimeException e) {
                stop(connector.getArn());
                throw e;
            }
        } catch (IOException e) {
            throw new IllegalStateException("Unable to prepare Kafka Connect plugin archive", e);
        }
    }
    public void configure(KafkaConnectConnector connector) {
        Worker worker = workers.get(connector.getArn());
        if (worker == null) {
            return;
        }
        IOException lastFailure = null;
        for (int attempt = 0; attempt < 30; attempt++) {
            try {
                HttpURLConnection connection = (HttpURLConnection) URI.create(worker.endpoint() + "/connectors/" + connector.getName() + "/config")
                        .toURL().openConnection();
                connection.setRequestMethod("PUT");
                connection.setDoOutput(true);
                connection.setConnectTimeout(5000);
                connection.setReadTimeout(10000);
                connection.setRequestProperty("Content-Type", "application/json");
                objectMapper.writeValue(connection.getOutputStream(), connector.getConnectorConfiguration());
                int responseCode = connection.getResponseCode();
                if (responseCode == 404) {
                    throw new IOException("Kafka Connect worker has not registered its connector API yet");
                }
                if (responseCode >= 300) {
                    throw new IllegalStateException("Kafka Connect rejected connector configuration: " + responseCode);
                }
                return;
            } catch (IOException e) {
                lastFailure = e;
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException interrupted) {
                    Thread.currentThread().interrupt();
                    throw new IllegalStateException("Interrupted while waiting for Kafka Connect worker", interrupted);
                }
            }
        }
        throw new IllegalStateException("Kafka Connect worker is not ready for connector " + connector.getName(), lastFailure);
    }
    public void stop(String connectorArn) {
        Worker worker = workers.remove(connectorArn);
        if (worker == null) {
            lifecycleManager.removeIfExists(containerName(connectorArn));
            return;
        }
        lifecycleManager.stopAndRemove(worker.containerId(), worker.logs());
        try {
            Files.walk(worker.directory()).sorted(java.util.Comparator.reverseOrder()).forEach(path -> {
                try {
                    Files.deleteIfExists(path);
                } catch (IOException ignored) {
                    // A failed cleanup only leaves a disposable plugin extraction directory.
                }
            });
        } catch (IOException e) {
            LOG.debugv("Could not remove Kafka Connect work directory {0}: {1}", worker.directory(), e.getMessage());
        }
    }
    private String distributedProperties(String bootstrap, String extra) {
        return "bootstrap.servers=" + bootstrap + "\n"
                + "group.id=floci-kafka-connect\nconfig.storage.topic=floci-connect-configs\n"
                + "offset.storage.topic=floci-connect-offsets\nstatus.storage.topic=floci-connect-status\n"
                + "config.storage.replication.factor=1\noffset.storage.replication.factor=1\n"
                + "status.storage.replication.factor=1\nkey.converter=org.apache.kafka.connect.json.JsonConverter\n"
                + "value.converter=org.apache.kafka.connect.json.JsonConverter\nplugin.path=/opt/floci/plugins\n"
                + (extra == null ? "" : extra + "\n");
    }
    private String containerName(String connectorArn) {
        return "floci-kafkaconnect-" + connectorArn.substring(connectorArn.lastIndexOf('/') + 1);
    }
    private void extractZip(byte[] archive, Path output) throws IOException {
        try (ZipInputStream zip = new ZipInputStream(new java.io.ByteArrayInputStream(archive))) {
            java.util.zip.ZipEntry entry;
            while ((entry = zip.getNextEntry()) != null) {
                Path target = output.resolve(entry.getName()).normalize();
                if (!target.startsWith(output)) {
                    throw new IOException("Plugin ZIP contains an invalid path");
                }
                if (entry.isDirectory()) { Files.createDirectories(target); }
                else { Files.createDirectories(target.getParent()); Files.copy(zip, target); }
            }
        }
    }
    private void copyWorkDirectory(String containerId, Path directory) throws IOException {
        ByteArrayOutputStream archive = new ByteArrayOutputStream();
        try (TarArchiveOutputStream tar = new TarArchiveOutputStream(archive);
             java.util.stream.Stream<Path> paths = Files.walk(directory)) {
            tar.setLongFileMode(TarArchiveOutputStream.LONGFILE_POSIX);
            tar.setBigNumberMode(TarArchiveOutputStream.BIGNUMBER_POSIX);
            for (Path path : paths.sorted(Comparator.naturalOrder()).toList()) {
                String relative = directory.relativize(path).toString();
                if (relative.isEmpty()) {
                    continue;
                }
                TarArchiveEntry entry = new TarArchiveEntry(path.toFile(), "floci/" + relative);
                tar.putArchiveEntry(entry);
                if (Files.isRegularFile(path)) {
                    Files.copy(path, tar);
                }
                tar.closeArchiveEntry();
            }
        }
        lifecycleManager.getDockerClient().copyArchiveToContainerCmd(containerId)
                .withRemotePath("/opt")
                .withTarInputStream(new ByteArrayInputStream(archive.toByteArray()))
                .exec();
    }
    private record Worker(String containerId, String endpoint, Path directory, Closeable logs) { }
 }
@@ -0,0 +1,41 @@
 package io.github.hectorvent.floci.services.kafkaconnect.model;
 import io.quarkus.runtime.annotations.RegisterForReflection;
 import java.time.Instant;
 import java.util.LinkedHashMap;
 import java.util.Map;
@RegisterForReflection
 public class KafkaConnectConnector {
    private String arn;
    private String name;
    private String state;
    private String clusterArn;
    private String pluginArn;
    private String workerConfigurationArn;
    private long currentVersion;
    private Instant creationTime;
    private Map<String, String> connectorConfiguration = new LinkedHashMap<>();
    private Map<String, String> tags = new LinkedHashMap<>();
    public String getArn() { return arn; }
    public void setArn(String arn) { this.arn = arn; }
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public String getState() { return state; }
    public void setState(String state) { this.state = state; }
    public String getClusterArn() { return clusterArn; }
    public void setClusterArn(String clusterArn) { this.clusterArn = clusterArn; }
    public String getPluginArn() { return pluginArn; }
    public void setPluginArn(String pluginArn) { this.pluginArn = pluginArn; }
    public String getWorkerConfigurationArn() { return workerConfigurationArn; }
    public void setWorkerConfigurationArn(String workerConfigurationArn) { this.workerConfigurationArn = workerConfigurationArn; }
    public long getCurrentVersion() { return currentVersion; }
    public void setCurrentVersion(long currentVersion) { this.currentVersion = currentVersion; }
    public Instant getCreationTime() { return creationTime; }
    public void setCreationTime(Instant creationTime) { this.creationTime = creationTime; }
    public Map<String, String> getConnectorConfiguration() { return connectorConfiguration; }
    public void setConnectorConfiguration(Map<String, String> connectorConfiguration) { this.connectorConfiguration = connectorConfiguration; }
    public Map<String, String> getTags() { return tags; }
    public void setTags(Map<String, String> tags) { this.tags = tags; }
 }
@@ -0,0 +1,32 @@
 package io.github.hectorvent.floci.services.kafkaconnect.model;
 import io.quarkus.runtime.annotations.RegisterForReflection;
 import java.time.Instant;
 import java.util.LinkedHashMap;
 import java.util.Map;
@RegisterForReflection
 public class KafkaConnectCustomPlugin {
    private String arn;
    private String name;
    private String contentType;
    private String bucketArn;
    private String fileKey;
    private Instant creationTime;
    private Map<String, String> tags = new LinkedHashMap<>();
    public String getArn() { return arn; }
    public void setArn(String arn) { this.arn = arn; }
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public String getContentType() { return contentType; }
    public void setContentType(String contentType) { this.contentType = contentType; }
    public String getBucketArn() { return bucketArn; }
    public void setBucketArn(String bucketArn) { this.bucketArn = bucketArn; }
    public String getFileKey() { return fileKey; }
    public void setFileKey(String fileKey) { this.fileKey = fileKey; }
    public Instant getCreationTime() { return creationTime; }
    public void setCreationTime(Instant creationTime) { this.creationTime = creationTime; }
    public Map<String, String> getTags() { return tags; }
    public void setTags(Map<String, String> tags) { this.tags = tags; }
 }
@@ -0,0 +1,24 @@
 package io.github.hectorvent.floci.services.kafkaconnect.model;
 import io.quarkus.runtime.annotations.RegisterForReflection;
 import java.time.Instant;
@RegisterForReflection
 public class KafkaConnectWorkerConfiguration {
    private String arn;
    private String name;
    private String propertiesFileContent;
    private long revision;
    private Instant creationTime;
    public String getArn() { return arn; }
    public void setArn(String arn) { this.arn = arn; }
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public String getPropertiesFileContent() { return propertiesFileContent; }
    public void setPropertiesFileContent(String propertiesFileContent) { this.propertiesFileContent = propertiesFileContent; }
    public long getRevision() { return revision; }
    public void setRevision(long revision) { this.revision = revision; }
    public Instant getCreationTime() { return creationTime; }
    public void setCreationTime(Instant creationTime) { this.creationTime = creationTime; }
 }
@@ -110,7 +110,11 @@ public class KinesisJsonHandler {
    private Response handleCreateStream(JsonNode request, String region) {
        String streamName = request.path("StreamName").asText();
-        int shardCount = request.path("ShardCount").asInt(1);
+        JsonNode shardCountNode = request.path("ShardCount");
        boolean hasShardCount = !shardCountNode.isMissingNode() && !shardCountNode.isNull();
        if (hasShardCount && (!shardCountNode.isIntegralNumber() || !shardCountNode.canConvertToInt())) {
            throw new AwsException("InvalidArgumentException", "ShardCount must be an integer.", 400);
        }
        String streamMode = null;
        JsonNode modeDetails = request.path("StreamModeDetails");
        if (modeDetails.isObject()) {
@@ -119,6 +123,18 @@ public class KinesisJsonHandler {
                streamMode = mode;
            }
        }
        if (hasShardCount && streamMode != null) {
            throw new AwsException("InvalidArgumentException",
                    "Specify either ShardCount or StreamModeDetails, not both.", 400);
        }
        if (!hasShardCount && streamMode == null) {
            throw new AwsException("InvalidArgumentException",
                    "Either ShardCount or StreamModeDetails must be specified.", 400);
        }
        if (hasShardCount && shardCountNode.intValue() < 1) {
            throw new AwsException("InvalidArgumentException", "ShardCount must be at least 1.", 400);
        }
        int shardCount = hasShardCount ? shardCountNode.intValue() : 1;
        // CreateStream's optional Tags member was being dropped. That is invisible to a
        // hand-written script but not to Terraform: aws_kinesis_stream sets tags at create
        // time and then reads them back with ListTagsForStream, so an empty read produces a
@@ -633,6 +649,15 @@ public class KinesisJsonHandler {
        String shardId = request.path("ShardId").asText();
        String type = request.path("ShardIteratorType").asText();
        String seq = request.has("StartingSequenceNumber") ? request.path("StartingSequenceNumber").asText() : null;
        if (!Set.of("AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "AT_TIMESTAMP", "LATEST", "TRIM_HORIZON")
                .contains(type)) {
            throw new AwsException("InvalidArgumentException", "Invalid ShardIteratorType: " + type, 400);
        }
        if (("AT_SEQUENCE_NUMBER".equals(type) || "AFTER_SEQUENCE_NUMBER".equals(type))
                && (seq == null || seq.isBlank())) {
            throw new AwsException("InvalidArgumentException",
                    "ShardIteratorType " + type + " requires a StartingSequenceNumber", 400);
        }
        // AWS sends Timestamp as epoch seconds (double with fractional ms).
        // Convert to long millis at the boundary; the emulator stores time in ms everywhere.
        // Use Math.round to avoid 1ms drift from FP multiplication (e.g. X.999...).
@@ -661,6 +686,9 @@ public class KinesisJsonHandler {
    private Response handleGetRecords(JsonNode request, String region) {
        String iterator = request.path("ShardIterator").asText();
        Integer limit = request.has("Limit") ? request.path("Limit").asInt() : null;
        if (limit != null && (limit < 1 || limit > 10000)) {
            throw new AwsException("InvalidArgumentException", "Limit must be between 1 and 10000.", 400);
        }
        Map<String, Object> result = service.getRecords(iterator, limit, region);
        List<KinesisRecord> records = (List<KinesisRecord>) result.get("Records");
@@ -906,7 +906,7 @@ public class KinesisService implements ResourceProvider {
            }
        }
-        int max = limit != null ? Math.min(limit, 1000) : 1000;
+        int max = limit != null ? Math.min(limit, 10000) : 10000;
        List<KinesisRecord> result = new ArrayList<>();
        int nextIndex = startIndex;
        for (int i = startIndex; i < allRecords.size() && result.size() < max; i++) {
@@ -68,25 +68,43 @@ public class ImageCacheService {
                LOG.infov("Image already present locally, skipping pull: {0}", imageUri);
                return resolvedImage;
            }
-            LOG.infov("Pulling image: {0}", imageUri);
+            return pullImage(imageUri, requestedPlatform, explicitPlatform, imageKey);
-            try {
+        }
-                runWithRetry(imageUri, MAX_PULL_ATTEMPTS, INITIAL_BACKOFF_MS, () -> {
+    }
-                    PullImageCmd pullImage = dockerClient.pullImageCmd(imageUri)
+
-                            .withAuthConfig(resolveAuth(imageUri));
+    public String refreshImage(String imageUri) {
-                    if (explicitPlatform) {
+        return refreshImage(imageUri, null);
-                        pullImage.withPlatform(requestedPlatform);
+    }
-                    }
+
-                    pullImage.exec(new PullImageResultCallback())
+    public String refreshImage(String imageUri, String platform) {
-                            .awaitCompletion(5, TimeUnit.MINUTES);
+        boolean explicitPlatform = platform != null && !platform.isBlank();
-                });
+        String requestedPlatform = explicitPlatform ? platform.trim() : daemonPlatform();
-                resolvedImage = resolvedImageReference(imageUri, inspectLocalImage(imageUri));
+        ImageKey imageKey = new ImageKey(imageUri, requestedPlatform);
-                resolvedImages.put(imageKey, resolvedImage);
+        Object lock = locks.computeIfAbsent(imageUri, k -> new Object());
-                LOG.infov("Image pulled successfully: {0}", imageUri);
+        synchronized (lock) {
-                return resolvedImage;
+            return pullImage(imageUri, requestedPlatform, explicitPlatform, imageKey);
-            } catch (InterruptedException e) {
+        }
-                Thread.currentThread().interrupt();
+    }
-                throw new RuntimeException("Interrupted while pulling image: " + imageUri, e);
+
-            }
+    private String pullImage(String imageUri, String requestedPlatform, boolean explicitPlatform, ImageKey imageKey) {
        LOG.infov("Pulling image: {0}", imageUri);
        try {
            runWithRetry(imageUri, MAX_PULL_ATTEMPTS, INITIAL_BACKOFF_MS, () -> {
                PullImageCmd pullImage = dockerClient.pullImageCmd(imageUri)
                        .withAuthConfig(resolveAuth(imageUri));
                if (explicitPlatform) {
                    pullImage.withPlatform(requestedPlatform);
                }
                pullImage.exec(new PullImageResultCallback())
                        .awaitCompletion(5, TimeUnit.MINUTES);
            });
            String resolvedImage = resolvedImageReference(imageUri, inspectLocalImage(imageUri));
            resolvedImages.put(imageKey, resolvedImage);
            LOG.infov("Image pulled successfully: {0}", imageUri);
            return resolvedImage;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Interrupted while pulling image: " + imageUri, e);
        }
    }
@@ -91,8 +91,15 @@ public class MskController {
    @GET
    @Path("/v1/clusters/{clusterArn}/bootstrap-brokers")
-    public Response getBootstrapBrokers(@PathParam("clusterArn") String clusterArn) {
+    public Response getBootstrapBrokers(@PathParam("clusterArn") String clusterArn,
                                        @HeaderParam("X-Floci-Internal-Client") String internalClient) {
        String bootstrapBrokers = mskService.getBootstrapBrokers(clusterArn);
        if ("true".equals(internalClient)) {
            MskCluster cluster = mskService.describeCluster(clusterArn);
            if (cluster.getInternalBootstrapBrokers() != null) {
                bootstrapBrokers = cluster.getInternalBootstrapBrokers();
            }
        }
        return Response.ok(Map.of("bootstrapBrokerString", bootstrapBrokers)).build();
    }
@@ -7,7 +7,6 @@ import io.github.hectorvent.floci.core.common.docker.ContainerBuilder;
 import io.github.hectorvent.floci.core.common.docker.ContainerDetector;
 import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager;
 import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager.ContainerInfo;
 import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager.EndpointInfo;
 import io.github.hectorvent.floci.core.common.docker.ContainerLogStreamer;
 import io.github.hectorvent.floci.core.common.docker.ContainerSpec;
 import io.github.hectorvent.floci.core.common.docker.ContainerStorageHelper;
@@ -32,6 +31,7 @@ public class RedpandaManager {
    private static final Logger LOG = Logger.getLogger(RedpandaManager.class);
    private static final int KAFKA_PORT = 9092;
    private static final int INTERNAL_KAFKA_PORT = 9093;
    static final int ADMIN_PORT = 9644;
    // Redpanda's Admin API exposes readiness at /v1/status/ready. /ready always returns 404
@@ -124,28 +124,22 @@ public class RedpandaManager {
        // reachable address up front. The host port has to be known before the
        // container starts (the flag is a startup arg), so in native mode we
        // pre-allocate it ourselves instead of letting Docker assign one dynamically.
-        String kafkaAdvertiseAddr;
+        int kafkaHostPort = portAllocator.allocate(
-        int kafkaHostPort = 0;
+                config.services().msk().kafkaHostPortBase(),
-        if (!containerDetector.isRunningInContainer()) {
+                config.services().msk().kafkaHostPortMax());
-            kafkaHostPort = portAllocator.allocate(
+        String externalKafkaAdvertiseAddr = "external://localhost:" + kafkaHostPort;
-                    config.services().msk().kafkaHostPortBase(),
+        String internalKafkaAdvertiseAddr = "internal://" + containerName + ":" + INTERNAL_KAFKA_PORT;
                    config.services().msk().kafkaHostPortMax());
            kafkaAdvertiseAddr = "localhost:" + kafkaHostPort;
        } else {
            // Sibling containers on the docker network resolve the broker via its
            // container name through Docker's embedded DNS.
            kafkaAdvertiseAddr = containerName + ":" + KAFKA_PORT;
        }
        // Build command
        List<String> cmd = new ArrayList<>(List.of(
                "redpanda", "start", "--overprovisioned", "--smp", "1",
                "--memory", "512M", "--reserve-memory", "0M",
-                "--advertise-kafka-addr", kafkaAdvertiseAddr));
+                "--kafka-addr", "internal://0.0.0.0:" + INTERNAL_KAFKA_PORT + ",external://0.0.0.0:" + KAFKA_PORT,
                "--advertise-kafka-addr", internalKafkaAdvertiseAddr + "," + externalKafkaAdvertiseAddr));
-        // Build container spec. Publish Kafka/admin ports to the host only in
+        // MSK clients run outside Floci's Docker network as often as inside it. Publish and
-        // native mode; in Docker mode producers/consumers reach the broker via
+        // advertise the Kafka listener in both topologies so GetBootstrapBrokers is usable by
-        // the docker network IP resolved from container inspect.
+        // real host clients, not merely sibling containers.
        ContainerBuilder.Builder specBuilder = containerBuilder.newContainer(image)
                .withName(containerName)
                .withDockerNetwork(config.services().dockerNetwork())
@@ -155,11 +149,7 @@ public class RedpandaManager {
                        AwsArnUtils.accountOrDefault(cluster.getClusterArn(), regionResolver.getAccountId()),
                        AwsArnUtils.regionOrDefault(cluster.getClusterArn(), regionResolver.getDefaultRegion())));
-        if (!containerDetector.isRunningInContainer()) {
+        specBuilder.withPortBinding(KAFKA_PORT, kafkaHostPort).withDynamicPort(ADMIN_PORT);
            specBuilder.withPortBinding(KAFKA_PORT, kafkaHostPort).withDynamicPort(ADMIN_PORT);
        } else {
            specBuilder.withExposedPort(KAFKA_PORT).withExposedPort(ADMIN_PORT);
        }
        // Handle persistence mounting
        if (ContainerStorageHelper.isNamedVolumeMode(config)) {
@@ -183,17 +173,14 @@ public class RedpandaManager {
        try {
            info = lifecycleManager.createAndStart(spec);
        } catch (RuntimeException e) {
-            if (!containerDetector.isRunningInContainer()) {
+            portAllocator.release(kafkaHostPort);
                portAllocator.release(kafkaHostPort);
            }
            throw e;
        }
        cluster.setContainerId(info.containerId());
        // Resolve endpoints
-        EndpointInfo kafkaEndpoint = info.getEndpoint(KAFKA_PORT);
+        cluster.setBootstrapBrokers("localhost:" + kafkaHostPort);
-
+        cluster.setInternalBootstrapBrokers(containerName + ":" + INTERNAL_KAFKA_PORT);
        cluster.setBootstrapBrokers(kafkaEndpoint.host() + ":" + kafkaEndpoint.port());
        LOG.infov("Redpanda container {0} started. Bootstrap: {1}", info.containerId(), cluster.getBootstrapBrokers());
        // Attach log streaming (new feature)
@@ -231,7 +218,13 @@ public class RedpandaManager {
                return false;
            }
        } else {
-            String containerIp = bootstrap.split(":")[0];
+            String containerIp = lifecycleManager.getDockerClient().inspectContainerCmd(cluster.getContainerId())
                    .exec().getNetworkSettings().getNetworks().values().stream()
                    .map(network -> network.getIpAddress()).filter(ip -> ip != null && !ip.isBlank())
                    .findFirst().orElse(null);
            if (containerIp == null) {
                return false;
            }
            adminUrl = "http://" + containerIp + ":" + ADMIN_PORT + ADMIN_READY_PATH;
        }
@@ -89,6 +89,10 @@ public class MskCluster {
    @JsonProperty("bootstrapBrokers")
    private String bootstrapBrokers;
    // Broker address for Floci services sharing the Docker network.
    @JsonProperty("internalBootstrapBrokers")
    private String internalBootstrapBrokers;
    // Docker container ID for mock=false
    @JsonProperty("containerId")
    private String containerId;
@@ -178,6 +182,11 @@ public class MskCluster {
    public String getBootstrapBrokers() { return bootstrapBrokers; }
    public void setBootstrapBrokers(String bootstrapBrokers) { this.bootstrapBrokers = bootstrapBrokers; }
    public String getInternalBootstrapBrokers() { return internalBootstrapBrokers; }
    public void setInternalBootstrapBrokers(String internalBootstrapBrokers) {
        this.internalBootstrapBrokers = internalBootstrapBrokers;
    }
    public String getContainerId() { return containerId; }
    public void setContainerId(String containerId) { this.containerId = containerId; }
@@ -155,6 +155,8 @@ floci:
    services:
      ssm:
        flush-interval-ms: 5000
      kafkaconnect:
        flush-interval-ms: 5000
      dynamodb:
        flush-interval-ms: 5000
      sns:
@@ -357,6 +359,10 @@ floci:
      default-image: "redpandadata/redpanda:latest"
      kafka-host-port-base: 9300
      kafka-host-port-max: 9399
    kafkaconnect:
      enabled: true
      mock: false
      worker-image: apache/kafka:3.9.0
    amazonmq:
      enabled: true
      mock: false
@@ -35,13 +35,14 @@ images:
  - imageId: ami-0abcdef1234567892
    aliases:
      - ami-ubuntu2004
-    dockerImage: public.ecr.aws/docker/library/ubuntu:20.04
+    dockerImage: floci/ami-ubuntu:20.04
    name: ubuntu/images/hvm-ssd/ubuntu-focal-20.04-amd64-server-20230324
    description: Canonical, Ubuntu, 20.04 LTS
    ownerId: "099720109477"
    imageOwnerAlias: canonical
    architecture: x86_64
    creationDate: "2023-03-24T00:00:00.000Z"
    guestRuntime: systemd
  - imageId: ami-ubuntu2204
    dockerImage: public.ecr.aws/docker/library/ubuntu:22.04
@@ -95,6 +96,14 @@ images:
    architecture: x86_64
    creationDate: "2023-04-11T00:00:00.000Z"
  - imageId: ami-debian13
    dockerImage: floci/ami-debian:13
    name: debian-13-amd64
    description: Debian 13 AMI with SSH
    architecture: x86_64
    creationDate: "2026-10-06T00:00:00.000Z"
    guestRuntime: systemd
  - imageId: ami-alpine
    dockerImage: public.ecr.aws/docker/library/alpine:latest
    name: alpine-linux-latest-x86_64
@@ -103,14 +112,15 @@ images:
    creationDate: "2023-04-01T00:00:00.000Z"
  - imageId: ami-fedora42
-    dockerImage: registry.fedoraproject.org/fedora:42
+    dockerImage: floci/ami-fedora:42
    name: Fedora-Cloud-Base-42-x86_64
    description: Fedora Linux 42 AMI
    architecture: x86_64
    creationDate: "2025-04-15T00:00:00.000Z"
    guestRuntime: systemd
  - imageId: ami-opensuse-leap156
-    dockerImage: registry.opensuse.org/opensuse/leap:15.6
+    dockerImage: floci/ami-opensuse:15.6
    name: openSUSE-Leap-15.6-x86_64
    description: openSUSE Leap 15.6 AMI
    architecture: x86_64
@@ -1,6 +1,7 @@
 package io.github.hectorvent.floci.lifecycle;
 import io.github.hectorvent.floci.config.EmulatorConfig;
 import io.github.hectorvent.floci.services.ec2.Ec2Service;
 import io.github.hectorvent.floci.config.FlociCertificateAuthority;
 import io.github.hectorvent.floci.core.common.ContainerTeardown;
 import io.github.hectorvent.floci.core.common.Resettable;
@@ -36,6 +37,7 @@ class EmulatorInfoControllerTest {
    @Mock private Instance<ContainerTeardown> containerTeardowns;
    @Mock private FlociCertificateAuthority certificateAuthority;
    @Mock private EmulatorConfig config;
    @Mock private Ec2Service ec2Service;
    @Mock private ContainerTeardown sageMakerTeardown;
    @Mock private ContainerTeardown batchTeardown;
    @Mock private Resettable resettable;
@@ -45,7 +47,7 @@ class EmulatorInfoControllerTest {
    @BeforeEach
    void setUp() {
        controller = new EmulatorInfoController(serviceRegistry, initLifecycleState, storageFactory,
-                resettables, containerTeardowns, certificateAuthority, config);
+                resettables, containerTeardowns, certificateAuthority, config, ec2Service);
    }
    @Test
@@ -115,6 +115,6 @@ class CloudWatchLogsMetricFilterResetTest {
        when(resettables.iterator()).thenAnswer(call -> List.of(resettable).iterator());
        Instance<ContainerTeardown> teardowns = mock(Instance.class);
        when(teardowns.iterator()).thenAnswer(call -> List.<ContainerTeardown>of().iterator());
-        return new EmulatorInfoController(null, null, f.factory, resettables, teardowns, null, null);
+        return new EmulatorInfoController(null, null, f.factory, resettables, teardowns, null, null, null);
    }
 }
@@ -281,7 +281,7 @@ class CodeDeployEcsIntegrationTest {
    @Test
    @Order(11)
-    void deploymentCompletesSuccessfully() throws InterruptedException {
+    void deploymentFailsWithoutRealGreenTargets() throws InterruptedException {
        Assumptions.assumeTrue(deploymentId != null, "deploymentId must be set");
        String status = null;
@@ -302,7 +302,7 @@ class CodeDeployEcsIntegrationTest {
            TimeUnit.MILLISECONDS.sleep(500);
        }
-        Assertions.assertEquals("Succeeded", status, "ECS deployment should succeed");
+        Assertions.assertEquals("Failed", status, "ECS deployment without green targets must fail");
    }
    @Test
@@ -350,14 +350,14 @@ class CodeDeployEcsIntegrationTest {
        .then()
            .statusCode(200)
            .body("deploymentTargets[0].deploymentTargetType", equalTo("ECSTarget"))
-            .body("deploymentTargets[0].ecsTarget.status", equalTo("Succeeded"));
+            .body("deploymentTargets[0].ecsTarget.status", equalTo("Failed"));
    }
    @Test
    @Order(14)
-    void listenerNowPointsToGreen() {
+    void listenerRemainsOnBlueAfterFailedDeployment() {
        Assumptions.assumeTrue(listenerArn != null, "listenerArn must be set");
-        Assumptions.assumeTrue(greenTgArn != null, "greenTgArn must be set");
+        Assumptions.assumeTrue(blueTgArn != null, "blueTgArn must be set");
        given()
            .contentType(ELBV2_CONTENT_TYPE)
@@ -369,6 +369,6 @@ class CodeDeployEcsIntegrationTest {
        .then()
            .statusCode(200)
            .body("DescribeRulesResponse.DescribeRulesResult.Rules.member.Actions.member.TargetGroupArn",
-                    equalTo(greenTgArn));
+                    equalTo(blueTgArn));
    }
 }
@@ -8,7 +8,15 @@ import io.github.hectorvent.floci.services.ec2.Ec2Service;
 import io.github.hectorvent.floci.services.ec2.model.Instance;
 import io.github.hectorvent.floci.services.ec2.model.Reservation;
 import io.github.hectorvent.floci.services.ecs.EcsService;
 import io.github.hectorvent.floci.services.ecs.model.EcsLoadBalancer;
 import io.github.hectorvent.floci.services.ecs.model.EcsServiceModel;
 import io.github.hectorvent.floci.services.ecs.model.LaunchType;
 import io.github.hectorvent.floci.services.ecs.model.TaskSet;
 import io.github.hectorvent.floci.services.elbv2.ElbV2Service;
 import io.github.hectorvent.floci.services.elbv2.model.Action;
 import io.github.hectorvent.floci.services.elbv2.model.Listener;
 import io.github.hectorvent.floci.services.elbv2.model.Rule;
 import io.github.hectorvent.floci.services.elbv2.model.TargetHealth;
 import io.github.hectorvent.floci.services.lambda.LambdaService;
 import io.github.hectorvent.floci.services.lambda.model.LambdaAlias;
 import io.github.hectorvent.floci.services.lambda.model.LambdaFunction;
@@ -32,6 +40,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -494,6 +503,247 @@ class CodeDeployServiceTest {
        verify(elbV2Service).shiftListenerForward(REGION, "listener", "blue", "green", 100);
    }
    @Test
    void ecsGreenTaskSetWaitsForFargateTasksToStart() throws InterruptedException {
        EcsService ecsService = mock(EcsService.class);
        AtomicInteger sleeps = new AtomicInteger();
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService,
                mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                ignored -> sleeps.incrementAndGet());
        TaskSet pending = greenTaskSet();
        pending.setRunningCount(0);
        TaskSet running = greenTaskSet();
        when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION))
                .thenReturn(List.of(pending), List.of(running));
        TaskSet actual = ecsDeploymentService.waitForGreenTaskSetRunning(REGION, "cluster", "service", pending,
                new java.util.concurrent.atomic.AtomicBoolean());
        assertEquals(running, actual);
        assertEquals(1, sleeps.get());
    }
    @Test
    void ecsGreenTaskSetFailsAfterFargateStartupTimeout() {
        EcsService ecsService = mock(EcsService.class);
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService,
                mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                ignored -> { });
        TaskSet pending = greenTaskSet();
        pending.setRunningCount(0);
        when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(pending));
        IllegalStateException exception = assertThrows(IllegalStateException.class,
                () -> ecsDeploymentService.waitForGreenTaskSetRunning(REGION, "cluster", "service", pending,
                        new java.util.concurrent.atomic.AtomicBoolean()));
        assertEquals("ECS green task set has no running tasks", exception.getMessage());
    }
    @Test
    void ecsGreenTaskSetMustMatchServiceLaunchTypeAndTargetGroup() {
        EcsService ecsService = mock(EcsService.class);
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService,
                mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                Thread::sleep);
        TaskSet green = greenTaskSet();
        green.setLaunchType(LaunchType.EC2);
        when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(green));
        assertThrows(IllegalStateException.class, () -> ecsDeploymentService.verifyGreenTaskSet(REGION,
                "cluster", "service", "deployment", fargateService(), green, "green-target-group"));
    }
    @Test
    void ecsGreenTaskSetAcceptsTheExistingEc2Path() {
        EcsService ecsService = mock(EcsService.class);
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService,
                mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                Thread::sleep);
        TaskSet green = greenTaskSet();
        green.setLaunchType(LaunchType.EC2);
        EcsServiceModel ec2Service = new EcsServiceModel();
        ec2Service.setLaunchType(LaunchType.EC2);
        when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(green));
        ecsDeploymentService.verifyGreenTaskSet(REGION, "cluster", "service", "deployment", ec2Service,
                green, "green-target-group");
    }
    @Test
    void ecsGreenTaskSetRejectsEc2WithoutRunningTasksImmediately() {
        EcsService ecsService = mock(EcsService.class);
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService,
                mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                ignored -> { });
        TaskSet green = greenTaskSet();
        green.setLaunchType(LaunchType.EC2);
        green.setRunningCount(0);
        EcsServiceModel ec2Service = new EcsServiceModel();
        ec2Service.setLaunchType(LaunchType.EC2);
        when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(green));
        IllegalStateException exception = assertThrows(IllegalStateException.class,
                () -> ecsDeploymentService.verifyGreenTaskSet(REGION, "cluster", "service", "deployment",
                        ec2Service, green, "green-target-group"));
        assertEquals("ECS green task set has no running tasks", exception.getMessage());
    }
    @Test
    void ecsDeploymentVerifiesTheAuthoritativeDefaultRuleInsteadOfStaleListenerActions() {
        ElbV2Service elbV2Service = mock(ElbV2Service.class);
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class),
                elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                Thread::sleep);
        Listener staleListener = new Listener();
        staleListener.setDefaultActions(List.of(forwardAction("blue-target-group")));
        when(elbV2Service.describeListeners(REGION, null, List.of("listener"))).thenReturn(List.of(staleListener));
        when(elbV2Service.describeRules(REGION, "listener", null))
                .thenReturn(List.of(defaultRule(forwardAction("green-target-group"))));
        ecsDeploymentService.verifyListenersForwardToGreen(REGION, List.of("listener"), "green-target-group");
        verify(elbV2Service, never()).describeListeners(anyString(), any(), any());
    }
    @Test
    void ecsDeploymentAcceptsWeightedDefaultRuleWithAllTrafficOnGreen() {
        ElbV2Service elbV2Service = mock(ElbV2Service.class);
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class),
                elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                Thread::sleep);
        Action weightedForward = new Action();
        weightedForward.setType("forward");
        weightedForward.setTargetGroups(List.of(targetGroup("blue-target-group", 0),
                targetGroup("green-target-group", 100)));
        when(elbV2Service.describeRules(REGION, "listener", null))
                .thenReturn(List.of(defaultRule(weightedForward)));
        ecsDeploymentService.verifyListenersForwardToGreen(REGION, List.of("listener"), "green-target-group");
    }
    @Test
    void ecsDeploymentRejectsDefaultRuleWithResidualBlueTraffic() {
        ElbV2Service elbV2Service = mock(ElbV2Service.class);
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class),
                elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                Thread::sleep);
        Action weightedForward = new Action();
        weightedForward.setType("forward");
        weightedForward.setTargetGroups(List.of(targetGroup("blue-target-group", 1),
                targetGroup("green-target-group", 99)));
        when(elbV2Service.describeRules(REGION, "listener", null))
                .thenReturn(List.of(defaultRule(weightedForward)));
        IllegalStateException exception = assertThrows(IllegalStateException.class,
                () -> ecsDeploymentService.verifyListenersForwardToGreen(REGION,
                        List.of("listener"), "green-target-group"));
        assertEquals("Production listener did not switch to the green target group", exception.getMessage());
    }
    @Test
    void ecsDeploymentRejectsAnEmptyGreenTargetGroup() {
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class),
                mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                ignored -> { });
        assertThrows(IllegalStateException.class, () -> ecsDeploymentService.waitForGreenTargets(REGION,
                "green-target-group", new java.util.concurrent.atomic.AtomicBoolean()));
    }
    @Test
    void ecsDeploymentWaitsForHealthyGreenTargets() throws InterruptedException {
        ElbV2Service elbV2Service = mock(ElbV2Service.class);
        AtomicInteger sleeps = new AtomicInteger();
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class),
                elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                ignored -> sleeps.incrementAndGet());
        TargetHealth target = new TargetHealth();
        target.setState("healthy");
        when(elbV2Service.describeTargetHealth(REGION, "green-target-group", null))
                .thenReturn(List.of(), List.of(target));
        ecsDeploymentService.waitForGreenTargets(REGION, "green-target-group",
                new java.util.concurrent.atomic.AtomicBoolean());
        assertEquals(1, sleeps.get());
    }
    @Test
    void ecsDeploymentWaitsForDefaultTargetGroupHealthThreshold() throws InterruptedException {
        ElbV2Service elbV2Service = mock(ElbV2Service.class);
        AtomicInteger polls = new AtomicInteger();
        CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class),
                elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(),
                new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null,
                ignored -> { });
        TargetHealth initial = new TargetHealth();
        initial.setState("initial");
        TargetHealth healthy = new TargetHealth();
        healthy.setState("healthy");
        when(elbV2Service.describeTargetHealth(REGION, "green-target-group", null))
                .thenAnswer(ignored -> List.of(polls.incrementAndGet() > 150 ? healthy : initial));
        ecsDeploymentService.waitForGreenTargets(REGION, "green-target-group",
                new java.util.concurrent.atomic.AtomicBoolean());
        assertEquals(151, polls.get());
    }
    private TaskSet greenTaskSet() {
        EcsLoadBalancer loadBalancer = new EcsLoadBalancer();
        loadBalancer.setTargetGroupArn("green-target-group");
        TaskSet taskSet = new TaskSet();
        taskSet.setTaskSetArn("green");
        taskSet.setExternalId("deployment");
        taskSet.setStartedBy("CODE_DEPLOY");
        taskSet.setStatus("ACTIVE");
        taskSet.setLaunchType(LaunchType.FARGATE);
        taskSet.setComputedDesiredCount(1);
        taskSet.setRunningCount(1);
        taskSet.setLoadBalancers(List.of(loadBalancer));
        return taskSet;
    }
    private EcsServiceModel fargateService() {
        EcsServiceModel ecsService = new EcsServiceModel();
        ecsService.setLaunchType(LaunchType.FARGATE);
        return ecsService;
    }
    private Rule defaultRule(Action action) {
        Rule rule = new Rule();
        rule.setDefault(true);
        rule.setActions(List.of(action));
        return rule;
    }
    private Action forwardAction(String targetGroupArn) {
        Action action = new Action();
        action.setType("forward");
        action.setTargetGroupArn(targetGroupArn);
        return action;
    }
    private Action.TargetGroupTuple targetGroup(String targetGroupArn, int weight) {
        Action.TargetGroupTuple target = new Action.TargetGroupTuple();
        target.setTargetGroupArn(targetGroupArn);
        target.setWeight(weight);
        return target;
    }
    // ---- Helpers --------------------------------------------------------------------------
    private String createServerDeployment(String appName, String groupName, int timeoutSeconds) {
@@ -394,14 +394,17 @@ class CodePipelineServiceTest {
    }
    @Test
-    void codeDeployToEcsRegistersTaskDefinitionAndSubstitutesAppSpec() throws Exception {
+    void codeDeployToEcsSubstitutesFargateTaskDefinitionImageFromImageDetail() throws Exception {
        S3Service s3Service = mock(S3Service.class);
        CodeDeployService codeDeployService = mock(CodeDeployService.class);
        EcsService ecsService = mock(EcsService.class);
        S3Object source = new S3Object();
        source.setData(zipArtifact(Map.of(
                "taskdef.json", """
-                        {"family":"pipeline-task","containerDefinitions":[{"name":"app","image":"nginx"}]}
+                        {"family":"pipeline-task","networkMode":"awsvpc","requiresCompatibilities":["FARGATE"],"containerDefinitions":[{"name":"app","image":"<IMAGE1_NAME>"}]}
                        """,
                "imageDetail.json", """
                        {"ImageURI":"000000000000.dkr.ecr.us-east-1.amazonaws.com/pipeline-app:build-5"}
                        """,
                "appspec.json", """
                        {"version":0.0,"Resources":[{"TargetService":{"Type":"AWS::ECS::Service","Properties":{"TaskDefinition":"<TASK_DEFINITION>","LoadBalancerInfo":{"ContainerName":"app","ContainerPort":80}}}}]}
@@ -429,6 +432,9 @@ class CodePipelineServiceTest {
            ArgumentCaptor<RegisterTaskDefinitionRequest> request = ArgumentCaptor.forClass(RegisterTaskDefinitionRequest.class);
            verify(ecsService).registerTaskDefinition(request.capture(), eq(REGION));
            assertEquals("pipeline-task", request.getValue().getFamily());
            assertTrue(request.getValue().isFargate());
            assertEquals("000000000000.dkr.ecr.us-east-1.amazonaws.com/pipeline-app:build-5",
                    request.getValue().getContainerDefinitions().getFirst().getImage());
            ArgumentCaptor<Map<String, Object>> revision = ArgumentCaptor.forClass(Map.class);
            verify(codeDeployService).createDeployment(eq(REGION), eq("ecs-app"), eq("ecs-group"),
                    eq(null), revision.capture(), anyString());
@@ -447,7 +453,8 @@ class CodePipelineServiceTest {
        EcsService ecsService = mock(EcsService.class);
        S3Object source = new S3Object();
        source.setData(zipArtifact(Map.of(
-                "taskdef.json", "{\"family\":\"pipeline-task\",\"containerDefinitions\":[]}",
+                "taskdef.json", "{\"family\":\"pipeline-task\",\"containerDefinitions\":[{\"name\":\"app\",\"image\":\"<IMAGE1_NAME>\"}]}",
                "imageDetail.json", "{\"ImageURI\":\"000000000000.dkr.ecr.us-east-1.amazonaws.com/pipeline-app:build-5\"}",
                "appspec.json", "{\"version\":0.0,\"Resources\":[{\"TargetService\":{\"Properties\":{\"TaskDefinition\":\"<TASK_DEFINITION>\"}}}]}")));
        source.setETag("source-etag");
        source.setLastModified(Instant.now());
@@ -964,8 +971,12 @@ class CodePipelineServiceTest {
                                "configuration": {
                                    "ApplicationName": "ecs-app",
                                    "DeploymentGroupName": "ecs-group",
-                                    "TaskDefinitionTemplatePath": "SourceOutput::taskdef.json",
+                                    "TaskDefinitionTemplateArtifact": "SourceOutput",
-                                    "AppSpecTemplatePath": "SourceOutput::appspec.json"
+                                    "TaskDefinitionTemplatePath": "taskdef.json",
                                    "AppSpecTemplateArtifact": "SourceOutput",
                                    "AppSpecTemplatePath": "appspec.json",
                                    "Image1ArtifactName": "SourceOutput",
                                    "Image1ContainerName": "app"
                                },
                                "inputArtifacts": [{"name": "SourceOutput"}]
                            }]
@@ -854,6 +854,14 @@ class Ec2ContainerManagerTest {
                "dnf must be probed before yum");
    }
    @Test
    void sshdInstallProbeHandlesOpenSuse() {
        String script = Ec2ContainerManager.sshdInstallProbeCommand()[2];
        assertTrue(script.contains("command -v zypper"), script);
        assertTrue(script.contains("zypper --non-interactive install -y openssh"), script);
    }
    @Test
    void metadataProxyInstallCommandHandlesYumForAmazonLinux2() {
        // The IMDS proxy dependency install had the same gap, and unlike the sshd probe it
@@ -866,6 +874,15 @@ class Ec2ContainerManagerTest {
        assertTrue(script.contains("yum install -y iproute iptables socat curl ca-certificates"), script);
    }
    @Test
    void metadataProxyInstallCommandHandlesOpenSuse() {
        String script = Ec2ContainerManager.metadataProxyInstallCommand()[2];
        assertTrue(script.contains("command -v zypper"), script);
        assertTrue(script.contains("zypper --non-interactive install -y iproute2 iptables socat curl ca-certificates"),
                script);
    }
    @Test
    void metadataProxyInstallCommandLetsDnfReplaceCurlMinimalWithCurl() {
        // public.ecr.aws/amazonlinux/amazonlinux:2023 -- the fallback image AmiImageResolver
@@ -1354,9 +1371,10 @@ class Ec2ContainerManagerTest {
    private static final String[] SSHD_INSTALL_PROBE_CMD = {"sh", "-c",
            "if ! command -v sshd >/dev/null 2>&1 || ! command -v scp >/dev/null 2>&1; then"
            + "  if command -v dnf >/dev/null 2>&1; then dnf install -y openssh-server openssh-clients >/dev/null 2>&1;"
-            + "  elif command -v yum >/dev/null 2>&1; then yum install -y openssh-server openssh-clients >/dev/null 2>&1;"
+             + "  elif command -v yum >/dev/null 2>&1; then yum install -y openssh-server openssh-clients >/dev/null 2>&1;"
-            + "  elif command -v apt-get >/dev/null 2>&1; then DEBIAN_FRONTEND=noninteractive apt-get install -y openssh-server openssh-client >/dev/null 2>&1;"
+             + "  elif command -v apt-get >/dev/null 2>&1; then DEBIAN_FRONTEND=noninteractive apt-get install -y openssh-server openssh-client >/dev/null 2>&1;"
-            + "  elif command -v apk >/dev/null 2>&1; then apk add --no-cache openssh >/dev/null 2>&1;"
+             + "  elif command -v zypper >/dev/null 2>&1; then zypper --non-interactive install -y openssh >/dev/null 2>&1;"
             + "  elif command -v apk >/dev/null 2>&1; then apk add --no-cache openssh >/dev/null 2>&1;"
            + "  fi;"
            + "fi;"
            + "command -v sshd >/dev/null 2>&1 || exit 1;"
@@ -37,6 +37,7 @@ class Ec2ImageCatalogTest {
                "ami-ubuntu2404-amd64",
                "ami-ubuntu2404-cloud-arm64",
                "ami-debian12",
                "ami-debian13",
                "ami-alpine",
                "ami-fedora42",
                "ami-opensuse-leap156"), imageIds);
@@ -78,9 +79,14 @@ class Ec2ImageCatalogTest {
    }
    @Test
-    void resolverUsesFedoraAndOpenSuseImages() {
+    void resolverUsesFedoraAndLocalSshReadyImages() {
-        assertEquals("registry.fedoraproject.org/fedora:42", amiImageResolver.resolve("ami-fedora42"));
+        assertEquals("floci/ami-ubuntu:20.04", amiImageResolver.resolve("ami-0abcdef1234567892"));
-        assertEquals("registry.opensuse.org/opensuse/leap:15.6", amiImageResolver.resolve("ami-opensuse-leap156"));
+        assertEquals("floci/ami-debian:13", amiImageResolver.resolve("ami-debian13"));
        assertEquals("floci/ami-fedora:42", amiImageResolver.resolve("ami-fedora42"));
        assertEquals("floci/ami-opensuse:15.6", amiImageResolver.resolve("ami-opensuse-leap156"));
        assertTrue(amiImageResolver.resolveImage("ami-0abcdef1234567892").systemd());
        assertTrue(amiImageResolver.resolveImage("ami-debian13").systemd());
        assertTrue(amiImageResolver.resolveImage("ami-fedora42").systemd());
    }
    @Test
@@ -2876,6 +2876,26 @@ class Ec2IntegrationTest {
            .extract().path("RunInstancesResponse.instancesSet.item.instanceId");
    }
    @Test
    @Order(80)
    void runInstancesUsesSecurityGroupsFromNetworkInterface() {
        given()
            .formParam("Action", "RunInstances")
            .formParam("ImageId", "ami-0abcdef1234567890")
            .formParam("InstanceType", "t2.micro")
            .formParam("MinCount", "1")
            .formParam("MaxCount", "1")
            .formParam("NetworkInterface.1.DeviceIndex", "0")
            .formParam("NetworkInterface.1.SubnetId", subnetId)
            .formParam("NetworkInterface.1.SecurityGroupId.1", securityGroupId)
            .header("Authorization", AUTH_HEADER)
        .when()
            .post("/")
        .then()
            .statusCode(200)
            .body("RunInstancesResponse.instancesSet.item.groupSet.item.groupId", equalTo(securityGroupId));
    }
    @Test
    @Order(80)
    void runInstancesWithArm64ImageDescribesArm64Architecture() {
@@ -108,6 +108,8 @@ class EcsContainerManagerEcrRewriteTest {
        List<String> images = imageCaptor.getAllValues();
        assertEquals(REWRITTEN_IMAGE, images.get(0), "ECR-shaped image should be rewritten for the docker pull");
        assertEquals("sidecar:latest", images.get(1), "non-ECR image should pass through unchanged");
        verify(lifecycleManager).refreshImage(REWRITTEN_IMAGE);
        verify(lifecycleManager).refreshImage("sidecar:latest");
        // RunTask/DescribeTasks must keep reporting the original task-def image, not the rewrite.
        List<Container> containers = task.getContainers();
@@ -300,6 +300,23 @@ class ElastiCacheIntegrationTest {
    @Test
    @Order(13)
    void keyExpiresAtTheConfiguredTtl() throws Exception {
        try (Socket socket = openSocket(firstProxyPort)) {
            write(socket, respArray("AUTH", GROUP_AUTH_TOKEN));
            assertEquals("+OK\r\n", readLine(socket));
            write(socket, respArray("SET", "expires-after-one-second", "value", "EX", "1"));
            assertEquals("+OK\r\n", readLine(socket));
            Thread.sleep(1_100);
            write(socket, respArray("GET", "expires-after-one-second"));
            assertEquals("$-1\r\n", readLine(socket));
        }
    }
    @Test
    @Order(14)
    void deleteUserRemovesUserFromDescribeUsers() {
        given()
            .formParam("Action", "DeleteUser")
@@ -322,7 +339,7 @@ class ElastiCacheIntegrationTest {
    }
    @Test
-    @Order(14)
+    @Order(15)
    void deleteReplicationGroupReleasesProxyPortForReuse() {
        given()
            .formParam("Action", "DeleteReplicationGroup")
@@ -225,6 +225,35 @@ class ElbV2ServiceTest {
                .get(rule.getRuleArn()).containsKey("env"));
    }
    @Test
    void shiftListenerForwardPersistsTheGreenTargetGroup() {
        SharedStorageFactory storageFactory = new SharedStorageFactory();
        ElbV2Service first = serviceWithStorage(storageFactory, mock(ElbV2DataPlane.class), mock(ElbV2HealthChecker.class));
        String lbArn = first.createLoadBalancer(
                REGION, "traffic-lb", "internal", "application", "ipv4",
                ALB_SUBNETS, List.of("sg-a"), Map.of()).getLoadBalancerArn();
        String blueTargetGroupArn = first.createTargetGroup(
                REGION, "traffic-blue", "HTTP", "HTTP1", 8080, "vpc-a", "ip",
                "HTTP", "traffic-port", true, "/", 30, 5, 5, 2, "200", "ipv4", Map.of()).getTargetGroupArn();
        String greenTargetGroupArn = first.createTargetGroup(
                REGION, "traffic-green", "HTTP", "HTTP1", 8080, "vpc-a", "ip",
                "HTTP", "traffic-port", true, "/", 30, 5, 5, 2, "200", "ipv4", Map.of()).getTargetGroupArn();
        String listenerArn = first.createListener(
                REGION, lbArn, "HTTP", 9080, null, List.of(),
                List.of(forwardAction(blueTargetGroupArn)), List.of(), Map.of()).getListenerArn();
        first.shiftListenerForward(REGION, listenerArn, blueTargetGroupArn, greenTargetGroupArn, 100);
        ElbV2Service reloaded = serviceWithStorage(storageFactory, mock(ElbV2DataPlane.class), mock(ElbV2HealthChecker.class));
        Rule defaultRule = reloaded.describeRules(REGION, listenerArn, null).stream()
                .filter(Rule::isDefault)
                .findFirst()
                .orElseThrow();
        assertEquals(greenTargetGroupArn, defaultRule.getActions().getFirst().getTargetGroupArn());
        assertEquals(greenTargetGroupArn, reloaded.describeListeners(REGION, lbArn, List.of(listenerArn)).getFirst()
                .getDefaultActions().getFirst().getTargetGroupArn());
    }
    @Test
    void initializeStorageDoesNotTouchCollaborators() {
        // #1913: ElbV2DataPlane.binding() calls back into this service, and during @PostConstruct the
@@ -0,0 +1,60 @@
 package io.github.hectorvent.floci.services.kafkaconnect;
 import io.github.hectorvent.floci.services.msk.MskService;
 import io.github.hectorvent.floci.services.msk.model.CreateClusterRequest;
 import io.github.hectorvent.floci.services.msk.model.MskCluster;
 import io.github.hectorvent.floci.services.s3.S3Service;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 import java.io.ByteArrayOutputStream;
 import java.util.Map;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipOutputStream;
 import static io.restassured.RestAssured.given;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.notNullValue;
@io.quarkus.test.junit.QuarkusTest
 class KafkaConnectIntegrationTest {
    @Inject S3Service s3Service;
    @Inject MskService mskService;
    @Test
    void customPluginWorkerConfigurationAndConnectorPersistThroughAwsRoutes() throws Exception {
        s3Service.createBucket("plugins", null);
        s3Service.putObject("plugins", "sink.zip", zip(), "application/zip", Map.of());
        CreateClusterRequest clusterRequest = new CreateClusterRequest();
        clusterRequest.setClusterName("connect-cluster");
        MskCluster cluster = mskService.createCluster(clusterRequest);
        String pluginArn = given().contentType("application/json")
                .body("""
                        {"customPluginName":"s3-sink","contentType":"ZIP","contentLocation":{"s3Location":{"bucketArn":"arn:aws:s3:::plugins","fileKey":"sink.zip"}}}
                        """)
                .when().post("/v1/custom-plugins").then().statusCode(200)
                .body("customPluginArn", notNullValue()).extract().path("customPluginArn");
        String workerArn = given().contentType("application/json")
                .body("{\"name\":\"worker\",\"propertiesFileContent\":\"offset.flush.interval.ms=1000\"}")
                .when().post("/v1/worker-configurations").then().statusCode(200)
                .body("revision", equalTo(1)).extract().path("workerConfigurationArn");
        given().contentType("application/json")
                .body("""
                        {"connectorName":"sink","connectorConfiguration":{"connector.class":"example.Sink","topics":"events"},"kafkaCluster":{"apacheKafkaCluster":{"bootstrapServers":["%s"]}},"plugins":[{"customPlugin":{"customPluginArn":"%s"}}],"workerConfiguration":{"workerConfigurationArn":"%s"}}
                        """.formatted(cluster.getBootstrapBrokers(), pluginArn, workerArn))
                .when().post("/v1/connectors").then().statusCode(200).body("connectorState", equalTo("RUNNING"));
        given().when().get("/v1/connectors").then().statusCode(200).body("connectors.size()", equalTo(1));
    }
    private byte[] zip() throws Exception {
        ByteArrayOutputStream output = new ByteArrayOutputStream();
        try (ZipOutputStream zip = new ZipOutputStream(output)) {
            zip.putNextEntry(new ZipEntry("README.txt"));
            zip.write("plugin".getBytes(java.nio.charset.StandardCharsets.UTF_8));
            zip.closeEntry();
        }
        return output.toByteArray();
    }
 }
@@ -584,7 +584,7 @@ class KinesisIntegrationTest {
    @Test
    @Order(31)
-    void createStreamWithOnDemandMode() {
+    void createStreamRejectsShardCountAndOnDemandModeTogether() {
        given()
            .header("X-Amz-Target", "Kinesis_20131202.CreateStream")
            .contentType(KINESIS_CONTENT_TYPE)
@@ -594,19 +594,8 @@ class KinesisIntegrationTest {
        .when()
            .post("/")
        .then()
-            .statusCode(200);
+            .statusCode(400)
-
+            .body("__type", equalTo("InvalidArgumentException"));
        given()
            .header("X-Amz-Target", "Kinesis_20131202.DescribeStream")
            .contentType(KINESIS_CONTENT_TYPE)
            .body("""
                {"StreamName": "on-demand-create-test"}
                """)
        .when()
            .post("/")
        .then()
            .statusCode(200)
            .body("StreamDescription.StreamModeDetails.StreamMode", equalTo("ON_DEMAND"));
    }
    @Test
@@ -339,7 +339,6 @@ class KinesisJsonHandlerTest {
    void createStreamHonorsOnDemandStreamMode() {
        ObjectNode req = MAPPER.createObjectNode();
        req.put("StreamName", "test-stream");
        req.put("ShardCount", 1);
        req.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND");
        assertThat(handler.handle("CreateStream", req, REGION).getStatus(), is(200));
@@ -1249,7 +1248,6 @@ class KinesisJsonHandlerTest {
    void updateMaxRecordSizeRejectsOnDemandStreams() {
        ObjectNode create = MAPPER.createObjectNode();
        create.put("StreamName", "on-demand-stream");
        create.put("ShardCount", 1);
        create.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND");
        assertThat(handler.handle("CreateStream", create, REGION).getStatus(), is(200));
@@ -1363,7 +1361,6 @@ class KinesisJsonHandlerTest {
    void updateShardCountRejectsOnDemandStream() {
        ObjectNode create = MAPPER.createObjectNode();
        create.put("StreamName", "on-demand-stream");
        create.put("ShardCount", 2);
        create.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND");
        assertThat(handler.handle("CreateStream", create, REGION).getStatus(), is(200));
@@ -1584,6 +1581,76 @@ class KinesisJsonHandlerTest {
        assertEquals(Map.of("Foo", "Bar"), service.listTagsForStream("test-stream", REGION));
    }
    @Test
    void createStreamRequiresExactlyOneCapacitySetting() {
        ObjectNode missingCapacity = MAPPER.createObjectNode();
        missingCapacity.put("StreamName", "missing-capacity");
        AwsException missingException = assertThrows(AwsException.class,
                () -> handler.handle("CreateStream", missingCapacity, REGION));
        assertEquals("InvalidArgumentException", missingException.getErrorCode());
        ObjectNode bothCapacitySettings = MAPPER.createObjectNode();
        bothCapacitySettings.put("StreamName", "both-capacity-settings");
        bothCapacitySettings.put("ShardCount", 1);
        bothCapacitySettings.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND");
        AwsException bothException = assertThrows(AwsException.class,
                () -> handler.handle("CreateStream", bothCapacitySettings, REGION));
        assertEquals("InvalidArgumentException", bothException.getErrorCode());
        ObjectNode zeroShards = MAPPER.createObjectNode();
        zeroShards.put("StreamName", "zero-shards");
        zeroShards.put("ShardCount", 0);
        AwsException zeroException = assertThrows(AwsException.class,
                () -> handler.handle("CreateStream", zeroShards, REGION));
        assertEquals("InvalidArgumentException", zeroException.getErrorCode());
    }
    @Test
    void getShardIteratorValidatesIteratorTypeAndSequenceNumber() {
        createStream("test-stream");
        ObjectNode invalidType = MAPPER.createObjectNode();
        invalidType.put("StreamName", "test-stream");
        invalidType.put("ShardId", FIRST_SHARD_ID);
        invalidType.put("ShardIteratorType", "INVALID");
        AwsException invalidTypeException = assertThrows(AwsException.class,
                () -> handler.handle("GetShardIterator", invalidType, REGION));
        assertEquals("InvalidArgumentException", invalidTypeException.getErrorCode());
        ObjectNode missingSequence = MAPPER.createObjectNode();
        missingSequence.put("StreamName", "test-stream");
        missingSequence.put("ShardId", FIRST_SHARD_ID);
        missingSequence.put("ShardIteratorType", "AT_SEQUENCE_NUMBER");
        AwsException missingSequenceException = assertThrows(AwsException.class,
                () -> handler.handle("GetShardIterator", missingSequence, REGION));
        assertEquals("InvalidArgumentException", missingSequenceException.getErrorCode());
    }
    @Test
    void getRecordsAcceptsKinesisMaximumLimitAndRejectsOutOfRangeValues() {
        createStream("test-stream");
        ObjectNode iteratorRequest = MAPPER.createObjectNode();
        iteratorRequest.put("StreamName", "test-stream");
        iteratorRequest.put("ShardId", FIRST_SHARD_ID);
        iteratorRequest.put("ShardIteratorType", "TRIM_HORIZON");
        String iterator = responseEntity(handler.handle("GetShardIterator", iteratorRequest, REGION))
                .get("ShardIterator").asText();
        ObjectNode maximumLimit = MAPPER.createObjectNode();
        maximumLimit.put("ShardIterator", iterator);
        maximumLimit.put("Limit", 10000);
        assertThat(handler.handle("GetRecords", maximumLimit, REGION).getStatus(), is(200));
        for (int invalidLimit : List.of(0, 10001)) {
            ObjectNode invalidLimitRequest = MAPPER.createObjectNode();
            invalidLimitRequest.put("ShardIterator", iterator);
            invalidLimitRequest.put("Limit", invalidLimit);
            AwsException exception = assertThrows(AwsException.class,
                    () -> handler.handle("GetRecords", invalidLimitRequest, REGION));
            assertEquals("InvalidArgumentException", exception.getErrorCode());
        }
    }
    /** Every non-string tag value shape — number, boolean, null, object, array — plus a map mixing a valid and an invalid value. */
    private static List<ObjectNode> nonStringTagValues() {
        ObjectNode number = MAPPER.createObjectNode();
@@ -284,6 +284,34 @@ class ImageCacheServiceTest {
        verify(pullImage).exec(any(PullImageResultCallback.class));
    }
    @Test
    void refreshesMutableTagWhileCachedImageStillExists() throws Exception {
        DockerClient dockerClient = mock(DockerClient.class);
        InspectImageCmd inspectImage = mock(InspectImageCmd.class);
        InspectImageCmd inspectCachedImage = mock(InspectImageCmd.class);
        PullImageCmd pullImage = mock(PullImageCmd.class);
        PullImageResultCallback callback = mock(PullImageResultCallback.class);
        when(dockerClient.inspectImageCmd(IMAGE)).thenReturn(inspectImage);
        when(dockerClient.inspectImageCmd("sha256:old")).thenReturn(inspectCachedImage);
        when(dockerClient.inspectImageCmd("sha256:new")).thenReturn(inspectCachedImage);
        when(inspectImage.exec()).thenReturn(
                new InspectImageResponse().withId("sha256:old").withOs("linux").withArch("amd64"),
                new InspectImageResponse().withId("sha256:new").withOs("linux").withArch("amd64"));
        when(inspectCachedImage.exec()).thenReturn(
                new InspectImageResponse().withId("sha256:old").withOs("linux").withArch("amd64"),
                new InspectImageResponse().withId("sha256:new").withOs("linux").withArch("amd64"));
        when(dockerClient.pullImageCmd(IMAGE)).thenReturn(pullImage);
        when(pullImage.withAuthConfig(any())).thenReturn(pullImage);
        when(pullImage.exec(any(PullImageResultCallback.class))).thenReturn(callback);
        ImageCacheService service = newService(dockerClient);
        assertEquals("sha256:old", service.ensureImageExists(IMAGE));
        assertEquals("sha256:new", service.refreshImage(IMAGE));
        assertEquals("sha256:new", service.ensureImageExists(IMAGE));
        verify(pullImage).exec(any(PullImageResultCallback.class));
    }
    @Test
    void succeedsOnFirstAttempt() throws Exception {
        AtomicInteger calls = new AtomicInteger();
@@ -43,7 +43,6 @@ import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -177,14 +176,15 @@ class RedpandaManagerTest {
        int flagIndex = spec.cmd().indexOf("--advertise-kafka-addr");
        assertTrue(flagIndex >= 0, "cmd should contain --advertise-kafka-addr");
-        assertEquals("localhost:54321", spec.cmd().get(flagIndex + 1));
+        assertEquals("internal://floci-msk-abc123:9093,external://localhost:54321", spec.cmd().get(flagIndex + 1));
        assertEquals(Integer.valueOf(54321), spec.portBindings().get(KAFKA_PORT));
    }
    @Test
-    void containerModeAdvertisesContainerNameAddress() {
+    void containerModeAdvertisesInternalAndExternalAddresses() {
        when(containerDetector.isRunningInContainer()).thenReturn(true);
        when(portAllocator.allocate(9300, 9399)).thenReturn(54321);
        ContainerInfo info = new ContainerInfo("container-456",
                Map.of(KAFKA_PORT, new EndpointInfo("172.18.0.5", KAFKA_PORT)));
@@ -199,15 +199,15 @@ class RedpandaManagerTest {
        int flagIndex = spec.cmd().indexOf("--advertise-kafka-addr");
        assertTrue(flagIndex >= 0, "cmd should contain --advertise-kafka-addr");
-        assertEquals("floci-msk-abc123:9092", spec.cmd().get(flagIndex + 1));
+        assertEquals("internal://floci-msk-abc123:9093,external://localhost:54321", spec.cmd().get(flagIndex + 1));
-        assertFalse(spec.portBindings().containsKey(KAFKA_PORT), "container mode should not publish ports to host");
+        assertEquals(Integer.valueOf(54321), spec.portBindings().get(KAFKA_PORT));
        assertTrue(spec.exposedPorts().contains(KAFKA_PORT));
        assertTrue(spec.exposedPorts().contains(RedpandaManager.ADMIN_PORT));
        assertFalse(Files.exists(tempDir.resolve("msk").resolve("test-cluster")),
                "container mode must not create host directories from inside the emulator container");
-        verifyNoInteractions(portAllocator);
+        verify(portAllocator).allocate(9300, 9399);
    }
    @Test
@@ -162,6 +162,10 @@ floci:
      enabled: true
      mock: true
      default-image: "redpandadata/redpanda:latest"
    kafkaconnect:
      enabled: true
      mock: true
      worker-image: apache/kafka:3.9.0
    amazonmq:
      enabled: true
      mock: true
@@ -10,6 +10,10 @@
 # the new-service sentinel only fires on genuinely unregistered handlers.
 services:
  - service: kafkaconnect
    doc: docs/services/kafka-connect.md
    sources:
      - { path: src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectController.java, mode: rest }
  - service: account
    doc: docs/services/account.md
    sources: