publishing-java-activity-events-recipe

Java Activity Event Publisher Recipe

Quick Reference

@Component
public class ActivityPublisherService {
    void publish(ActivityEvent event)                                    // Queue any event
    void publishActivity(String type, String entityId, Map<String,Object> meta) // Activity log
    void publishAuditLog(String action, String actorId, String resourceId)      // Audit trail
}

public record ActivityEvent(String type, String entityId, Map<String,Object> metadata, Instant occurredAt) {}

public interface ActivityHttpClient {  // Port — hexagonal outbound
    int post(ActivityEvent event);     // Returns HTTP status code
}

When to Use

• Activity/audit log publishing in Spring Boot services
• Non-blocking telemetry/observability clients
• Event-driven architectures with Java 21
• Async job/task submission
• Systems requiring retry resilience with virtual threads

Implementation Workflow (MANDATORY)

CRITICAL: Present plan to user BEFORE implementation

Phase 1: Planning (User Review Required)

Analyze requirements:

1. Event types → What events? (activity, audit, telemetry)
2. Volume → Events/second expected? (low <10, medium 10-1000, high >1000)
3. Destinations → Where? (single endpoint, multiple, fan-out)
4. Reliability → SLA? (best-effort, guaranteed, exactly-once)
5. Integration → Framework? (Spring Boot 3.2+, standalone)
6. JSON examples → Request examples of input events and expected output format

Ask user for JSON examples:

// Use AskUserQuestion tool FIRST to gather examples
questions = [{
  question: "Do you have example JSON for the events you want to publish?",
  header: "JSON Examples",
  options: [
    { label: "Yes - I'll provide examples",
      description: "Provide sample input event JSON and expected output format" },
    { label: "No - Use standard format",
      description: "Use Activity Stream or Audit Log standard formats" },
    { label: "Need help deciding",
      description: "Show me the standard formats first" }
  ],
  multiSelect: false
}]

Create implementation plan:

PLAN:
- Queue size → Based on volume (default 1000, adjust via @ConfigurationProperties)
- Retry policy → maxRetries, baseDelay based on SLA
- Worker pattern → Single virtual thread (default), multiple (high volume)
- Event types → publish() vs publishActivity() vs publishAuditLog()
- Error handling → Structured dead-letter logging, queue-full WARN
- Validation → Bean validation, required fields
- Testing strategy → Unit, integration (MockWebServer)

Present to user with AskUserQuestion:

questions = [{
  question: "Review the implementation plan for the activity publisher:",
  header: "Implementation Plan",
  options: [
    { label: "Approve - Proceed with implementation",
      description: "Queue: {queueSize}, Retry: {maxRetries}x, Volume: {volumeEstimate}" },
    { label: "Modify - Adjust parameters",
      description: "Change queue size, retry policy, or architecture decisions" },
    { label: "Explain - Need more details",
      description: "Provide more context about trade-offs and design decisions" }
  ],
  multiSelect: false
}]

User approval required → proceed to Phase 2

Phase 2: Implementation (After Approval)

File creation order (dependency-first):

Order	File	Purpose
1	ActivityEvent.java	Immutable record
2	ActivityHttpClient.java	Port interface
3	ActivityPublisherProperties.java	ConfigurationProperties
4	ActivityPublisherService.java	Core component
5	RestClientActivityHttpClient.java	Adapter (outbound)
6	ActivityPublisherServiceTest.java	Unit tests

Architecture Patterns

Event Record (Immutable)

public record ActivityEvent(
    String type,
    String entityId,
    Map<String, Object> metadata,
    Instant occurredAt
) {
    public static ActivityEvent of(String type, String entityId) {
        return new ActivityEvent(type, entityId, Map.of(), Instant.now());
    }
}

Port Interface (Hexagonal)

public interface ActivityHttpClient {
    int post(ActivityEvent event);  // Returns HTTP status code
}

Configuration Properties

@ConfigurationProperties("app.activity-publisher")
public record ActivityPublisherProperties(
    int queueSize,
    int maxRetries,
    Duration baseDelay
) {
    public ActivityPublisherProperties() {
        this(1000, 3, Duration.ofSeconds(1));
    }
}

Core Service (Virtual Thread Worker)

@Component
public class ActivityPublisherService {
    private final BlockingQueue<ActivityEvent> queue;
    private final ActivityHttpClient client;
    private final ActivityPublisherProperties props;
    private static final Logger log = LoggerFactory.getLogger(ActivityPublisherService.class);
    private Thread workerThread;

    public ActivityPublisherService(ActivityHttpClient client, ActivityPublisherProperties props) {
        this.client = client;
        this.props = props;
        this.queue = new LinkedBlockingQueue<>(props.queueSize());
    }

    @PostConstruct
    void start() {
        workerThread = Thread.ofVirtual().name("activity-publisher").start(this::processQueue);
    }

    @PreDestroy
    void stop() throws InterruptedException {
        workerThread.interrupt();
        workerThread.join(5_000);
    }

    public void publish(ActivityEvent event) {
        if (!queue.offer(event)) {
            log.warn("Activity queue full, dropping event: {}", event.type());
        }
    }

    public void publishActivity(String type, String entityId, Map<String, Object> metadata) {
        publish(new ActivityEvent(type, entityId, metadata, Instant.now()));
    }

    public void publishAuditLog(String action, String actorId, String resourceId) {
        publish(new ActivityEvent(
            "audit." + action,
            resourceId,
            Map.of("actorId", actorId, "action", action),
            Instant.now()
        ));
    }

    private void processQueue() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                ActivityEvent event = queue.poll(1, TimeUnit.SECONDS);
                if (event != null) dispatch(event);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    private void dispatch(ActivityEvent event) {
        int attempt = 0;
        while (attempt < props.maxRetries()) {
            try {
                int status = client.post(event);
                if (status >= 200 && status < 300) return;
                if (status >= 400 && status < 500 && status != 429 && status != 409) {
                    log.error("Dead-letter event type={} status={}", event.type(), status);
                    return;
                }
                // 429, 5xx, 409 → retry with backoff
            } catch (IOException e) {
                // Network error → retry with backoff
            }
            attempt++;
            if (attempt < props.maxRetries()) {
                long delayMs = props.baseDelay().toMillis() * (1L << attempt);
                try { Thread.sleep(delayMs); } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        }
        log.error("Max retries exceeded dead-letter event={}", event);
    }
}

HTTP Status Classification

Status	Category	Action
2xx	success	Return — remove from dispatch
429	rate_limited	Exponential backoff, retry
5xx, 409	retryable	Exponential backoff, retry
4xx (not 429, 409)	permanent_failure	Dead-letter log, drop
IOException	network_error	Exponential backoff, retry

Exponential Backoff

long delayMs = baseDelay.toMillis() * (1L << attempt);
// Default: 1s, 2s, 4s (attempt 1, 2, 3)

Outbound Adapter (RestClient)

@Component
public class RestClientActivityHttpClient implements ActivityHttpClient {
    private final RestClient restClient;

    public RestClientActivityHttpClient(RestClient.Builder builder,
                                        @Value("${app.activity-publisher.endpoint}") String endpoint) {
        this.restClient = builder.baseUrl(endpoint).build();
    }

    @Override
    public int post(ActivityEvent event) throws IOException {
        try {
            return restClient.post()
                .body(event)
                .retrieve()
                .toBodilessEntity()
                .getStatusCode()
                .value();
        } catch (RestClientException e) {
            throw new IOException("HTTP client error", e);
        }
    }
}

Error Handling

Scenario	Action
Queue full	log.warn → drop event → never blocks caller
Max retries exceeded	log.error structured JSON → dead-letter
Worker thread death	log.error → restart via @PostConstruct watchdog
IOException	Treat as retryable → backoff

Code Review Checklist

VALIDATE:
- [ ] publish() non-blocking? (queue.offer → immediate return)
- [ ] Bounded queue? (LinkedBlockingQueue with capacity)
- [ ] Queue full → log WARN + drop (never block)?
- [ ] Worker uses virtual thread (Thread.ofVirtual())?
- [ ] @PostConstruct starts worker, @PreDestroy interrupts + joins?
- [ ] Exponential backoff: baseDelay * 2^attempt?
- [ ] HTTP status classified correctly?
  - [ ] 2xx → success, return
  - [ ] 429, 5xx, 409 → retry with backoff
  - [ ] 4xx (not 429/409) → dead-letter, drop
  - [ ] IOException → retry with backoff
- [ ] ActivityHttpClient is an interface (port), not concrete class?
- [ ] @ConfigurationProperties used (not @Value on each field)?
- [ ] Dead-letter logging structured (log.error with fields, not System.out)?
- [ ] ActivityEvent is an immutable record?
- [ ] RestClient used (not RestTemplate)?
- [ ] Tests cover: success, 429 retry, 4xx dead-letter, queue-full drop?

Testing Patterns

Test	Assertion
Non-blocking	publish() returns < 5ms
Queue full	log.warn captured; event dropped
2xx success	client.post() called once; no retry
429 retry	client.post() called maxRetries times
4xx dead-letter	log.error captured; client.post() called once
IOException retry	client.post() called maxRetries times
@PreDestroy	workerThread.join completes within 6s

@ExtendWith(MockitoExtension.class)
class ActivityPublisherServiceTest {
    @Mock ActivityHttpClient client;
    ActivityPublisherService service;

    @BeforeEach void setUp() throws Exception {
        var props = new ActivityPublisherProperties(10, 3, Duration.ofMillis(10));
        service = new ActivityPublisherService(client, props);
        service.start();
    }

    @AfterEach void tearDown() throws Exception { service.stop(); }

    @Test void publish_queuesWithoutBlocking() {
        long start = System.currentTimeMillis();
        service.publish(ActivityEvent.of("test", "e1"));
        assertThat(System.currentTimeMillis() - start).isLessThan(5L);
    }
}

Validation Workflow

Phase	Action
1. SCOPE	Extract context: files, mode (informative | executive), sessionId
2. VERIFY	Run checklist against implementation
3. VIOLATIONS	Collect with file:line, check name, severity
4. REPORT	✓ Pass → Proceed | ✗ Fail → Return violations
5. METRICS	Call mcp__agent-orchestrator__store-skill-metrics
6. OUTPUT	Return JSON: violations[], fixRate, finalViolations

Metrics Structure:

{
  "sessionId": "string",
  "skill": "java-services:java-activity-publisher-recipe",
  "initialViolations": 0,
  "iterations": 1,
  "fixesApplied": 0,
  "finalViolations": 0,
  "mode": "informative | executive",
  "duration": 0.0
}

Output Format:

{
  "status": "success | failed",
  "violations": [
    {
      "file": "src/main/java/.../ActivityPublisherService.java",
      "line": 42,
      "check": "Non-blocking design",
      "violation": "publish() blocks on queue.put()",
      "severity": "critical"
    }
  ],
  "metrics": {
    "initialViolations": 2,
    "finalViolations": 0,
    "fixRate": 1.0
  }
}