OSAC-3428: Implement Provider Adapter framework with Kafka consumer lifecycle #174
+2,848
−237
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| version: "2" | ||
| run: | ||
| timeout: 5m | ||
| linters: | ||
| enable: | ||
| - errcheck | ||
| - govet | ||
| - ineffassign | ||
| - staticcheck | ||
| - unused | ||
| - misspell | ||
| - revive | ||
| settings: | ||
| revive: | ||
| rules: | ||
| - name: exported | ||
| disabled: true | ||
| formatters: | ||
| enable: | ||
| - goimports |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| # Copyright (c) 2026 Red Hat Inc. | ||
| # | ||
| # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except | ||
| # in compliance with the License. You may obtain a copy of the License at | ||
| # | ||
| # http://www.apache.org/licenses/LICENSE-2.0 | ||
|
|
||
| GINKGO ?= go run github.com/onsi/ginkgo/v2/ginkgo | ||
| GOLANGCI_LINT_VERSION ?= v2.12.1 | ||
|
|
||
| LOCALBIN ?= $(shell pwd)/bin | ||
| GOLANGCI_LINT ?= $(LOCALBIN)/golangci-lint | ||
|
|
||
| .PHONY: build-echo-adapter test lint clean | ||
|
|
||
| build-echo-adapter: | ||
| @mkdir -p bin | ||
| go build -o bin/echo-adapter ./cmd/echo-adapter | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| test: | ||
| $(GINKGO) run . | ||
|
|
||
| lint: $(GOLANGCI_LINT) | ||
| $(GOLANGCI_LINT) run . | ||
|
|
||
| $(GOLANGCI_LINT): | ||
| @mkdir -p $(LOCALBIN) | ||
| GOBIN=$(LOCALBIN) go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(GOLANGCI_LINT_VERSION) | ||
|
|
||
| clean: | ||
| rm -rf bin/ | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| /* | ||
| Copyright (c) 2026 Red Hat, Inc. | ||
|
|
||
| Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except | ||
| in compliance with the License. You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
|
|
||
| package adapters | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| cloudevents "github.com/cloudevents/sdk-go/v2" | ||
| ) | ||
|
|
||
| // MeteringEvent wraps a CloudEvent with its Kafka coordinates. | ||
| type MeteringEvent struct { | ||
| CloudEvent cloudevents.Event | ||
| Topic string | ||
| Partition int32 | ||
| Offset int64 | ||
| } | ||
|
|
||
| // SubmitResult is returned by Flush to report the outcome. | ||
| type SubmitResult struct { | ||
| ProviderEventID string | ||
| Idempotent bool | ||
| } | ||
|
|
||
| // ProviderAdapter is the interface that concrete provider adapters implement. | ||
| // The Runner calls Submit per event and Flush on a configurable interval. | ||
| type ProviderAdapter interface { | ||
| // Name returns the provider name used as a Prometheus label. | ||
| Name() string | ||
| // Submit processes a single metering event. | ||
| Submit(ctx context.Context, event MeteringEvent) error | ||
| // Flush uploads any buffered events. Called on the flush ticker | ||
| // (default 10s) and on graceful shutdown. | ||
| Flush(ctx context.Context) (SubmitResult, error) | ||
| // HealthCheck verifies connectivity to the provider. | ||
| HealthCheck(ctx context.Context) error | ||
| // Close releases resources after the final Flush. | ||
| Close() error | ||
|
Comment on lines
+44
to
+45
There was a problem hiding this comment. 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift Add a context to
Proposed interface change- Close() error
+ Close(ctx context.Context) errorAs per path instructions, use " 🤖 Prompt for AI AgentsSource: Path instructions
Contributor
Author
There was a problem hiding this comment. The current There was a problem hiding this comment.
However, the shutdown timeout on An internal timeout in each concrete adapter can mitigate this risk. The ✏️ Learnings added
You are interacting with an AI system. |
||
| } | ||
|
|
||
| // RetryableError is an optional marker for documentation purposes. The Runner | ||
| // retries all errors by default — only errors wrapped in NonRetryableError are | ||
| // skipped. Wrapping in RetryableError makes the intent explicit but does not | ||
| // change retry behavior. | ||
| type RetryableError struct{ Err error } | ||
|
|
||
| func (e *RetryableError) Error() string { return e.Err.Error() } | ||
| func (e *RetryableError) Unwrap() error { return e.Err } | ||
|
|
||
| // NonRetryableError signals the runner should skip the event without retry. | ||
| type NonRetryableError struct{ Err error } | ||
|
|
||
| func (e *NonRetryableError) Error() string { return e.Err.Error() } | ||
| func (e *NonRetryableError) Unwrap() error { return e.Err } | ||
|
|
||
| // KafkaConfig configures the Kafka consumer connection. | ||
| type KafkaConfig struct { | ||
| TLSEnabled bool // Enable TLS for broker connections | ||
| TLSCACert string // Path to CA certificate file (empty = system CAs) | ||
| SASLUser string // SASL/SCRAM username | ||
| SASLPassFile string // Path to file containing SASL password | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| /* | ||
| Copyright (c) 2026 Red Hat, Inc. | ||
|
|
||
| Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except | ||
| in compliance with the License. You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
|
|
||
| package adapters | ||
|
|
||
| import ( | ||
| "errors" | ||
|
|
||
| . "github.com/onsi/ginkgo/v2" | ||
| . "github.com/onsi/gomega" | ||
| ) | ||
|
|
||
| var _ = Describe("Error Types", func() { | ||
| Describe("RetryableError", func() { | ||
| It("wraps and unwraps the underlying error", func() { | ||
| underlying := errors.New("connection timeout") | ||
| err := &RetryableError{Err: underlying} | ||
|
|
||
| Expect(err.Error()).To(Equal("connection timeout")) | ||
| Expect(errors.Unwrap(err)).To(Equal(underlying)) | ||
| }) | ||
|
|
||
| It("can be detected with errors.As", func() { | ||
| underlying := errors.New("temporary failure") | ||
| err := &RetryableError{Err: underlying} | ||
|
|
||
| var retryable *RetryableError | ||
| Expect(errors.As(err, &retryable)).To(BeTrue()) | ||
| }) | ||
| }) | ||
|
|
||
| Describe("NonRetryableError", func() { | ||
| It("wraps and unwraps the underlying error", func() { | ||
| underlying := errors.New("malformed event") | ||
| err := &NonRetryableError{Err: underlying} | ||
|
|
||
| Expect(err.Error()).To(Equal("malformed event")) | ||
| Expect(errors.Unwrap(err)).To(Equal(underlying)) | ||
| }) | ||
|
|
||
| It("can be detected with errors.As", func() { | ||
| underlying := errors.New("permanent failure") | ||
| err := &NonRetryableError{Err: underlying} | ||
|
|
||
| var nonRetryable *NonRetryableError | ||
| Expect(errors.As(err, &nonRetryable)).To(BeTrue()) | ||
| }) | ||
|
|
||
| It("is not detected as RetryableError", func() { | ||
| err := &NonRetryableError{Err: errors.New("fail")} | ||
|
|
||
| var retryable *RetryableError | ||
| Expect(errors.As(err, &retryable)).To(BeFalse()) | ||
| }) | ||
| }) | ||
| }) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| /* | ||
| Copyright (c) 2026 Red Hat, Inc. | ||
|
|
||
| Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except | ||
| in compliance with the License. You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
|
|
||
| package adapters | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| . "github.com/onsi/ginkgo/v2" | ||
| . "github.com/onsi/gomega" | ||
| ) | ||
|
|
||
| func TestAdapters(t *testing.T) { | ||
| RegisterFailHandler(Fail) | ||
| RunSpecs(t, "Adapters Suite") | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
Repository: osac-project/osac
Length of output: 50374
🏁 Script executed:
Repository: osac-project/osac
Length of output: 25770
Remove AI tools from
Co-Authored-By.Use
Assisted-by/Generated-byfor AI assistance, and do not list AI tools such as Claude/Cursor inCo-Authored-By.🤖 Prompt for AI Agents
Source: Coding guidelines