diff --git a/docs/resources/function.md b/docs/resources/function.md index 242398c1..c1d38ae9 100644 --- a/docs/resources/function.md +++ b/docs/resources/function.md @@ -33,6 +33,7 @@ Manages Pulsar Functions through the Functions Worker API. - `disk_mb` (Number) The disk that need to be allocated per function instance - `forward_source_message_property` (Boolean) Whether to forward source message property to the function output message. - `go` (String) The path to the go file. +- `input_specs` (Block Set) Per-topic consumer configuration for the function's input topics, such as the receiver queue size. A topic configured here does not need to be repeated in `inputs`; if it is, this block takes precedence. (see [below for nested schema](#nestedblock--input_specs)) - `input_type_classname` (String) The input type class name of the function. - `inputs` (Set of String) The input topics of the function. - `jar` (String) The path to the jar file. @@ -62,6 +63,24 @@ Manages Pulsar Functions through the Functions Worker API. - `id` (String) The ID of this resource. + +### Nested Schema for `input_specs` + +Required: + +- `key` (String) The input topic that this consumer configuration applies to. + +Optional: + +- `consumer_properties` (Map of String) Consumer properties key/values for this topic. Pulsar 4.0.x does not return this field on read, so the provider preserves the configured value in state; import cannot recover existing consumer properties. +- `is_regex_pattern` (Boolean) Whether the topic is a regex pattern matching multiple topics. Cannot be changed in place; changing it replaces the function. +- `pool_messages` (Boolean) Whether the consumer pools messages for this topic. +- `receiver_queue_size` (Number) The consumer receiver queue size for this topic. Defaults to 1000, which buffers up to that many messages per function instance. Set to 0 to disable prefetch. +- `schema_properties` (Map of String) Schema properties key/values for this topic. +- `schema_type` (String) The schema type of this topic, either a builtin schema type such as `avro` or a Schema implementation class name. +- `serde_class_name` (String) The serde class name of this topic. Cannot be set together with `schema_type`. + + ### Nested Schema for `sink_config` diff --git a/go.mod b/go.mod index 7f3a8cb2..2789c9ec 100644 --- a/go.mod +++ b/go.mod @@ -3,14 +3,15 @@ module github.com/streamnative/terraform-provider-pulsar go 1.25.14 require ( - github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20260323144249-e7eb92f148f5 - github.com/cenkalti/backoff/v4 v4.2.1 + github.com/apache/pulsar-client-go v0.21.1-0.20260826023458-61d7a95e66cd + github.com/cenkalti/backoff/v4 v4.3.0 github.com/hashicorp/go-cty v1.4.1-0.20200414143053-d3edf31b6320 github.com/hashicorp/go-multierror v1.1.1 + github.com/hashicorp/terraform-plugin-go v0.19.0 github.com/hashicorp/terraform-plugin-log v0.9.0 github.com/hashicorp/terraform-plugin-sdk/v2 v2.30.0 github.com/pkg/errors v0.9.1 - github.com/stretchr/testify v1.10.0 + github.com/stretchr/testify v1.11.1 ) require ( @@ -35,7 +36,6 @@ require ( github.com/hashicorp/logutils v1.0.0 // indirect github.com/hashicorp/terraform-exec v0.19.0 // indirect github.com/hashicorp/terraform-json v0.17.1 // indirect - github.com/hashicorp/terraform-plugin-go v0.19.0 // indirect github.com/hashicorp/terraform-registry-address v0.2.2 // indirect github.com/hashicorp/terraform-svchost v0.1.1 // indirect github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d // indirect @@ -53,12 +53,12 @@ require ( github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect github.com/zclconf/go-cty v1.16.2 // indirect - golang.org/x/crypto v0.45.0 // indirect - golang.org/x/mod v0.29.0 // indirect - golang.org/x/net v0.47.0 // indirect + golang.org/x/crypto v0.48.0 // indirect + golang.org/x/mod v0.32.0 // indirect + golang.org/x/net v0.49.0 // indirect golang.org/x/oauth2 v0.28.0 // indirect - golang.org/x/sys v0.38.0 // indirect - golang.org/x/text v0.31.0 // indirect + golang.org/x/sys v0.42.0 // indirect + golang.org/x/text v0.34.0 // indirect google.golang.org/appengine v1.6.8 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20241015192408-796eee8c2d53 // indirect google.golang.org/grpc v1.69.4 // indirect diff --git a/go.sum b/go.sum index 11357004..b9859833 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk= -dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk= +dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= +dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= github.com/AthenZ/athenz v1.12.13 h1:OhZNqZsoBXNrKBJobeUUEirPDnwt0HRo4kQMIO1UwwQ= github.com/AthenZ/athenz v1.12.13/go.mod h1:XXDXXgaQzXaBXnJX6x/bH4yF6eon2lkyzQZ0z/dxprE= github.com/DataDog/zstd v1.5.0 h1:+K/VEwIAaPcHiMtQvpLD4lqW7f0Gk3xdYZmI1hD+CXo= @@ -14,14 +14,8 @@ github.com/acomagu/bufpipe v1.0.4 h1:e3H4WUzM3npvo5uv95QuJM3cQspFNtFBzvJ2oNjKIDQ github.com/acomagu/bufpipe v1.0.4/go.mod h1:mxdxdup/WdsKVreO5GpW4+M/1CE2sMG4jeGJ2sYmHc4= github.com/agext/levenshtein v1.2.2 h1:0S/Yg6LYmFJ5stwQeRp6EeOcCbj7xiqQSdNelsXvaqE= github.com/agext/levenshtein v1.2.2/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20251217021558-9d65d0ecf43f h1:L2zBKmcYD/qyko07XLxVYg97kXVqN+SMI/ZySf0m7+A= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20251217021558-9d65d0ecf43f/go.mod h1:GKmTD1u5YLuhUnoVTNGdhdGNAYhoglWNWgwLJZTljAw= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20260225054407-3564674615ac h1:h3oeTmmuYF6WwsAMFUMN5rvBbMwiwIY266LqZrQMTas= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20260225054407-3564674615ac/go.mod h1:/Zf8Q8bSSc6ndEJ8V1muIHf6ZWsMrHoQU+98Ww9pOeI= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20260311005904-7832a7da8515 h1:owdxfuv2bg2OncZu6jv4IWnHaQP/vSiqx4F5E52qLUI= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20260311005904-7832a7da8515/go.mod h1:/Zf8Q8bSSc6ndEJ8V1muIHf6ZWsMrHoQU+98Ww9pOeI= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20260323144249-e7eb92f148f5 h1:/PLxJWgzEAyyzuTcB/oOXAAkyuXolA1BYRSOPDZkVCY= -github.com/apache/pulsar-client-go v0.18.0-candidate-1.0.20260323144249-e7eb92f148f5/go.mod h1:/Zf8Q8bSSc6ndEJ8V1muIHf6ZWsMrHoQU+98Ww9pOeI= +github.com/apache/pulsar-client-go v0.21.1-0.20260826023458-61d7a95e66cd h1:e80eLDpEuha9xvjlDakKQfFh+Ronx+EM8mp3sVyhN3E= +github.com/apache/pulsar-client-go v0.21.1-0.20260826023458-61d7a95e66cd/go.mod h1:IDGs98WESBYof6In1rFOokCBx5UvyAiqSdWogXNIEo8= github.com/apparentlymart/go-textseg/v12 v12.0.0/go.mod h1:S/4uRK2UtaQttw1GenVJEynmyUenKwP++x/+DdGV/Ec= github.com/apparentlymart/go-textseg/v15 v15.0.0 h1:uYvfpb3DyLSCGWnctWKGj857c6ew1u1fNQOlOtuGxQY= github.com/apparentlymart/go-textseg/v15 v15.0.0/go.mod h1:K8XmNZdhEBkdlyDdvbmmsvpAG721bKi0joRfFdHIWJ4= @@ -33,8 +27,8 @@ github.com/bits-and-blooms/bitset v1.12.0 h1:U/q1fAF7xXRhFCrhROzIfffYnu+dlS38vCZ github.com/bits-and-blooms/bitset v1.12.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= github.com/bufbuild/protocompile v0.4.0 h1:LbFKd2XowZvQ/kajzguUp2DC9UEIQhIq77fZZlaQsNA= github.com/bufbuild/protocompile v0.4.0/go.mod h1:3v93+mbWn/v3xzN+31nwkJfrEpAUwp+BagBSZWx+TP8= -github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM= -github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8= @@ -62,8 +56,8 @@ github.com/go-git/go-git/v5 v5.9.0 h1:cD9SFA7sHVRdJ7AYck1ZaAa/yeuBvGPxwXDL8cxrOb github.com/go-git/go-git/v5 v5.9.0/go.mod h1:RKIqga24sWdMGZF+1Ekv9kylsDz6LzdTSI2s/OsZWE0= github.com/go-jose/go-jose/v4 v4.0.5 h1:M6T8+mKZl/+fNNuFHvGIzDz7BTLQPIounk/b9dw3AaE= github.com/go-jose/go-jose/v4 v4.0.5/go.mod h1:s3P1lRrkT8igV8D9OjyL4WRyHvjB6a4JSllnOrmmBOA= -github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= -github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-test/deep v1.0.3 h1:ZrJSEWsXzPOxaZnFteGEfooLba+ju3FYIbOrS+rQd68= @@ -143,8 +137,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kevinburke/ssh_config v1.2.0 h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4= github.com/kevinburke/ssh_config v1.2.0/go.mod h1:CT57kijsi8u/K/BOFA39wgDQJ9CxiF4nAY/ojJ6r6mM= -github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= +github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= @@ -208,16 +202,16 @@ github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0leargg github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= github.com/sergi/go-diff v1.2.0 h1:XU+rvMAioB0UC3q1MFrIQy4Vo5/4VsRDQQXHsEya6xQ= github.com/sergi/go-diff v1.2.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM= -github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= -github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/skeema/knownhosts v1.2.0 h1:h9r9cf0+u7wSE+M183ZtMGgOJKiL96brpaz5ekfJCpM= github.com/skeema/knownhosts v1.2.0/go.mod h1:g4fPeYpque7P0xefxtGzV81ihjC8sX2IqpAoNkjxbMo= github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.7.2/go.mod h1:R6va5+xMeoiuVRoj+gSkQ7d3FALtqAAGI1FQKckRals= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= -github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/vmihailenco/msgpack v3.3.3+incompatible/go.mod h1:fy3FlTQTDXWkZ7Bh6AcGMlsjHatGryHQYUTf1ShIgkk= github.com/vmihailenco/msgpack v4.0.4+incompatible h1:dSLoQfGFAo3F6OoNhwUmLwVgaUXK79GlxNBwueZn0xI= github.com/vmihailenco/msgpack v4.0.4+incompatible/go.mod h1:fy3FlTQTDXWkZ7Bh6AcGMlsjHatGryHQYUTf1ShIgkk= @@ -232,33 +226,33 @@ github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/zclconf/go-cty v1.16.2 h1:LAJSwc3v81IRBZyUVQDUdZ7hs3SYs9jv0eZJDWHD/70= github.com/zclconf/go-cty v1.16.2/go.mod h1:VvMs5i0vgZdhYawQNq5kePSpLAoz8u1xvZgrPIxfnZE= -go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= -go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= -go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY= -go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI= -go.opentelemetry.io/otel/metric v1.34.0 h1:+eTR3U0MyfWjRDhmFMxe2SsW64QrZ84AOhvqS7Y+PoQ= -go.opentelemetry.io/otel/metric v1.34.0/go.mod h1:CEDrp0fy2D0MvkXE+dPV7cMi8tWZwX3dmaIhwPOaqHE= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= +go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= +go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= +go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= go.opentelemetry.io/otel/sdk v1.31.0 h1:xLY3abVHYZ5HSfOg3l2E5LUj2Cwva5Y7yGxnSW9H5Gk= go.opentelemetry.io/otel/sdk v1.31.0/go.mod h1:TfRbMdhvxIIr/B2N2LQW2S5v9m3gOQ/08KsbbO5BPT0= go.opentelemetry.io/otel/sdk/metric v1.31.0 h1:i9hxxLJF/9kkvfHppyLL55aW7iIJz4JjxTeYusH7zMc= go.opentelemetry.io/otel/sdk/metric v1.31.0/go.mod h1:CRInTMVvNhUKgSAMbKyTMxqOBC0zgyxzW55lZzX43Y8= -go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k= -go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE= +go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= +go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= -golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= -golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= +golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= +golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= -golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA= -golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w= +golang.org/x/mod v0.32.0 h1:9F4d3PHLljb6x//jOyokMv3eX+YDeepZSEo3mFJy93c= +golang.org/x/mod v0.32.0/go.mod h1:SgipZ/3h2Ci89DlEtEXWUk/HteuRin+HHhN+WbNhguU= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= -golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= -golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= +golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= +golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= golang.org/x/oauth2 v0.28.0 h1:CrgCKl8PPAVtLnU3c+EDw6x11699EWlsDeWNWKdIOkc= golang.org/x/oauth2 v0.28.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -276,19 +270,19 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= -golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= -golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU= -golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254= +golang.org/x/term v0.40.0 h1:36e4zGLqU4yhjlmxEaagx2KuYbJq3EwY8K943ZsHcvg= +golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= -golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= -golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= diff --git a/pulsar/resource_pulsar_function.go b/pulsar/resource_pulsar_function.go index d510b6e4..cfb63651 100644 --- a/pulsar/resource_pulsar_function.go +++ b/pulsar/resource_pulsar_function.go @@ -43,6 +43,7 @@ const ( resourceFunctionGoKey = "go" resourceFunctionClassNameKey = "classname" resourceFunctionInputsKey = "inputs" + resourceFunctionInputSpecsKey = "input_specs" resourceFunctionTopicsPatternKey = "topics_pattern" resourceFunctionOutputKey = "output" resourceFunctionParallelismKey = "parallelism" @@ -79,6 +80,28 @@ const ( resourceFunctionRuntimeConfigConfigsKey = "configs" ) +// Attributes of a single `input_specs` block, mapping onto utils.ConsumerConfig. +const ( + resourceFunctionInputSpecTopicKey = "key" + resourceFunctionInputSpecReceiverQueueSizeKey = "receiver_queue_size" + resourceFunctionInputSpecSchemaTypeKey = "schema_type" + resourceFunctionInputSpecSerdeClassNameKey = "serde_class_name" + resourceFunctionInputSpecRegexPatternKey = "is_regex_pattern" + resourceFunctionInputSpecPoolMessagesKey = "pool_messages" + resourceFunctionInputSpecSchemaPropertiesKey = "schema_properties" + resourceFunctionInputSpecConsumerPropertiesKey = "consumer_properties" +) + +const defaultFunctionReceiverQueueSize = 1000 + +var functionInputSourceKeys = []string{ + resourceFunctionInputsKey, + resourceFunctionTopicsPatternKey, + resourceFunctionCustomSerdeInputsKey, + resourceFunctionCustomSchemaInputsKey, + resourceFunctionInputSpecsKey, +} + const ( runtimeOptionSinkConfigKey = "sinkConfig" runtimeOptionSourceConfigKey = "sourceConfig" @@ -122,6 +145,7 @@ func init() { resourceFunctionGoKey: "The path to the go file.", resourceFunctionClassNameKey: "The class name of the function.", resourceFunctionInputsKey: "The input topics of the function.", + resourceFunctionInputSpecsKey: "Per-topic consumer configuration for the function's input topics, such as the receiver queue size. A topic configured here does not need to be repeated in `inputs`; if it is, this block takes precedence.", resourceFunctionTopicsPatternKey: "The input topics pattern of the function. The pattern is a regex expression. The function consumes from all topics matching the pattern.", resourceFunctionOutputKey: "The output topic of the function.", resourceFunctionParallelismKey: "The parallelism of the function.", @@ -162,6 +186,7 @@ func resourcePulsarFunction() *schema.Resource { ReadContext: resourcePulsarFunctionRead, UpdateContext: resourcePulsarFunctionUpdate, DeleteContext: resourcePulsarFunctionDelete, + CustomizeDiff: resourcePulsarFunctionCustomizeDiff, Description: "Manages Pulsar Functions through the Functions Worker API.", Importer: &schema.ResourceImporter{ StateContext: func(ctx context.Context, d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) { @@ -222,14 +247,81 @@ func resourcePulsarFunction() *schema.Resource { resourceFunctionInputsKey: { Type: schema.TypeSet, Optional: true, - ForceNew: true, Description: resourceFunctionDescriptions[resourceFunctionInputsKey], Elem: &schema.Schema{Type: schema.TypeString}, }, + resourceFunctionInputSpecsKey: { + Type: schema.TypeSet, + Optional: true, + Description: resourceFunctionDescriptions[resourceFunctionInputSpecsKey], + // Note the deliberate absence of ForceNew on the nested attributes. Changing any + // of them rehashes the set element, which the SDK reads as a removal plus an + // addition, so a nested ForceNew would replace the function on the very edits + // Pulsar accepts in place. resourcePulsarFunctionCustomizeDiff decides + // replacement instead, using the same rule the broker enforces. + Elem: &schema.Resource{ + Schema: map[string]*schema.Schema{ + resourceFunctionInputSpecTopicKey: { + Type: schema.TypeString, + Required: true, + Description: "The input topic that this consumer configuration applies to.", + }, + resourceFunctionInputSpecReceiverQueueSizeKey: { + Type: schema.TypeInt, + Optional: true, + Default: defaultFunctionReceiverQueueSize, + //nolint:lll + Description: "The consumer receiver queue size for this topic. Defaults to 1000, which buffers up to that many messages per function instance. Set to 0 to disable prefetch.", + ValidateFunc: func(val interface{}, key string) ([]string, []error) { + if v := val.(int); v < 0 { + return nil, []error{ + fmt.Errorf("%s must be greater than or equal to 0, got %d", key, v), + } + } + return nil, nil + }, + }, + resourceFunctionInputSpecSchemaTypeKey: { + Type: schema.TypeString, + Optional: true, + //nolint:lll + Description: "The schema type of this topic, either a builtin schema type such as `avro` or a Schema implementation class name.", + }, + resourceFunctionInputSpecSerdeClassNameKey: { + Type: schema.TypeString, + Optional: true, + Description: "The serde class name of this topic. Cannot be set together with `schema_type`.", + }, + resourceFunctionInputSpecRegexPatternKey: { + Type: schema.TypeBool, + Optional: true, + //nolint:lll + Description: "Whether the topic is a regex pattern matching multiple topics. Cannot be changed in place; changing it replaces the function.", + }, + resourceFunctionInputSpecPoolMessagesKey: { + Type: schema.TypeBool, + Optional: true, + Description: "Whether the consumer pools messages for this topic.", + }, + resourceFunctionInputSpecSchemaPropertiesKey: { + Type: schema.TypeMap, + Optional: true, + Description: "Schema properties key/values for this topic.", + Elem: &schema.Schema{Type: schema.TypeString}, + }, + resourceFunctionInputSpecConsumerPropertiesKey: { + Type: schema.TypeMap, + Optional: true, + //nolint:lll + Description: "Consumer properties key/values for this topic. Pulsar 4.0.x does not return this field on read, so the provider preserves the configured value in state; import cannot recover existing consumer properties.", + Elem: &schema.Schema{Type: schema.TypeString}, + }, + }, + }, + }, resourceFunctionTopicsPatternKey: { Type: schema.TypeString, Optional: true, - ForceNew: true, Description: resourceFunctionDescriptions[resourceFunctionTopicsPatternKey], }, resourceFunctionOutputKey: { @@ -470,6 +562,341 @@ func resourcePulsarFunction() *schema.Resource { } } +// resourcePulsarFunctionCustomizeDiff validates input_specs and decides when a change to any input +// representation requires replacing the function. +// +// Pulsar's FunctionConfigUtils.validateUpdate() rejects any input topic the existing function does +// not already consume ("Input Topics cannot be altered") and any change to a topic's isRegexPattern +// flag, but accepts every other consumer setting - receiverQueueSize included. So replacement is +// keyed on the effective set of input topics rather than on any one schema attribute changing. +// +// The comparison includes every legacy input representation. Moving a topic from any legacy field +// into input_specs leaves that effective set untouched and must remain an in-place update. +func resourcePulsarFunctionCustomizeDiff(_ context.Context, diff *schema.ResourceDiff, _ interface{}) error { + newSpecs := diff.Get(resourceFunctionInputSpecsKey) + if err := validateFunctionInputSpecs(newSpecs); err != nil { + return err + } + + if diff.Id() == "" { + // On create there is nothing to replace. + return nil + } + + inputChanged := false + for _, key := range functionInputSourceKeys { + if diff.HasChange(key) { + inputChanged = true + break + } + } + if !inputChanged { + return nil + } + + oldInputs, newInputs := diff.GetChange(resourceFunctionInputsKey) + oldPattern, newPattern := diff.GetChange(resourceFunctionTopicsPatternKey) + oldCustomSerde, newCustomSerde := diff.GetChange(resourceFunctionCustomSerdeInputsKey) + oldCustomSchema, newCustomSchema := diff.GetChange(resourceFunctionCustomSchemaInputsKey) + oldSpecs, newSpecs := diff.GetChange(resourceFunctionInputSpecsKey) + + oldTopics := effectiveFunctionInputTopics( + oldInputs, oldPattern, oldCustomSerde, oldCustomSchema, oldSpecs, + ) + newTopics := effectiveFunctionInputTopics( + newInputs, newPattern, newCustomSerde, newCustomSchema, newSpecs, + ) + + if len(oldTopics) != len(newTopics) { + return forceNewFunctionInputTopology(diff, oldSpecs, newSpecs) + } + for topic, regexPattern := range newTopics { + oldRegexPattern, ok := oldTopics[topic] + if !ok || oldRegexPattern != regexPattern { + return forceNewFunctionInputTopology(diff, oldSpecs, newSpecs) + } + } + + return nil +} + +func validateFunctionInputSpecs(inputSpecs interface{}) error { + set, ok := inputSpecs.(*schema.Set) + if !ok || set.Len() == 0 { + return nil + } + + seenTopics := make(map[string]bool, set.Len()) + for _, item := range set.List() { + spec, ok := item.(map[string]interface{}) + if !ok { + continue + } + + topic, _ := spec[resourceFunctionInputSpecTopicKey].(string) + if topic == "" { + // The SDK can include an empty placeholder while diffing TypeSet elements. The nested + // Required schema validates user configuration, so ignore that internal value here. + continue + } + if seenTopics[topic] { + return fmt.Errorf("%s contains duplicate key %q", resourceFunctionInputSpecsKey, topic) + } + seenTopics[topic] = true + + schemaType, _ := spec[resourceFunctionInputSpecSchemaTypeKey].(string) + serdeClassName, _ := spec[resourceFunctionInputSpecSerdeClassNameKey].(string) + if schemaType != "" && serdeClassName != "" { + return fmt.Errorf("%s %q cannot set both %s and %s", + resourceFunctionInputSpecsKey, + topic, + resourceFunctionInputSpecSchemaTypeKey, + resourceFunctionInputSpecSerdeClassNameKey, + ) + } + } + + return nil +} + +func forceNewFunctionInputTopology(diff *schema.ResourceDiff, oldSpecs, newSpecs interface{}) error { + if diff.HasChange(resourceFunctionInputSpecsKey) { + return forceNewFunctionInputSpecs(diff, oldSpecs, newSpecs) + } + + if diff.HasChange(resourceFunctionInputsKey) { + return forceNewFunctionInputSet(diff, resourceFunctionInputsKey) + } + + if diff.HasChange(resourceFunctionTopicsPatternKey) { + return diff.ForceNew(resourceFunctionTopicsPatternKey) + } + + for _, key := range []string{ + resourceFunctionCustomSerdeInputsKey, + resourceFunctionCustomSchemaInputsKey, + } { + if diff.HasChange(key) { + return forceNewFunctionInputMap(diff, key) + } + } + + return errors.New("input topology changed without an input attribute diff") +} + +func forceNewFunctionInputSet(diff *schema.ResourceDiff, key string) error { + if err := diff.ForceNew(key); err != nil { + return err + } + + oldValue, newValue := diff.GetChange(key) + for _, value := range []interface{}{oldValue, newValue} { + set, ok := value.(*schema.Set) + if !ok { + continue + } + for _, item := range set.List() { + itemKey := fmt.Sprintf("%s.%d", key, set.F(item)) + if diff.HasChange(itemKey) { + return diff.ForceNew(itemKey) + } + } + } + + return nil +} + +func forceNewFunctionInputMap(diff *schema.ResourceDiff, key string) error { + if err := diff.ForceNew(key); err != nil { + return err + } + + oldValue, newValue := diff.GetChange(key) + mapKeys := functionInputMapKeys(oldValue) + for topic := range functionInputMapKeys(newValue) { + mapKeys[topic] = true + } + for topic := range mapKeys { + itemKey := key + "." + topic + if diff.HasChange(itemKey) { + return diff.ForceNew(itemKey) + } + } + + return nil +} + +// forceNewFunctionInputSpecs flags an input_specs change as requiring replacement. +// +// A set-level ForceNew is not enough on its own. schemaMap.diffSet only carries it into the diff +// through the "input_specs.#" count attribute, and it emits that attribute solely when the number of +// elements changes. Renaming a topic or flipping regex_pattern leaves the count identical, so the +// replacement would be silently dropped. Flag the nested attribute that actually changed as well, +// which does reach the diff. +func forceNewFunctionInputSpecs(diff *schema.ResourceDiff, oldSpecs, newSpecs interface{}) error { + if err := diff.ForceNew(resourceFunctionInputSpecsKey); err != nil { + return err + } + + for _, attribute := range []string{ + resourceFunctionInputSpecTopicKey, + resourceFunctionInputSpecRegexPatternKey, + } { + for _, specs := range []interface{}{oldSpecs, newSpecs} { + set, ok := specs.(*schema.Set) + if !ok { + continue + } + + for _, item := range set.List() { + key := fmt.Sprintf("%s.%d.%s", resourceFunctionInputSpecsKey, set.F(item), attribute) + if !diff.HasChange(key) { + continue + } + if err := diff.ForceNew(key); err != nil { + return err + } + } + } + } + + return nil +} + +// effectiveFunctionInputTopics maps every input topic the function consumes to its regex-pattern +// flag. Values are applied in Pulsar's create-path order, with input_specs last so it is the +// provider's canonical representation when a topic appears in more than one field. +func effectiveFunctionInputTopics( + inputs, topicsPattern, customSerdeInputs, customSchemaInputs, inputSpecs interface{}, +) map[string]bool { + topics := map[string]bool{} + + if set, ok := inputs.(*schema.Set); ok { + for _, item := range set.List() { + if topic, ok := item.(string); ok && topic != "" { + topics[topic] = false + } + } + } + + if pattern, ok := topicsPattern.(string); ok && pattern != "" { + topics[pattern] = true + } + + for topic := range functionInputMapKeys(customSerdeInputs) { + topics[topic] = false + } + for topic := range functionInputMapKeys(customSchemaInputs) { + topics[topic] = false + } + + // Applied last: an input_specs block wins over every legacy representation. + for topic, consumerConfig := range functionInputSpecsFromSchema(inputSpecs) { + topics[topic] = consumerConfig.RegexPattern + } + + return topics +} + +func functionInputMapKeys(value interface{}) map[string]bool { + keys := map[string]bool{} + + switch values := value.(type) { + case map[string]interface{}: + for key := range values { + if key != "" { + keys[key] = true + } + } + case map[string]string: + for key := range values { + if key != "" { + keys[key] = true + } + } + } + + return keys +} + +// functionInputSpecsFromSchema converts an input_specs set into the map shape the admin API expects. +func functionInputSpecsFromSchema(inputSpecs interface{}) map[string]utils.ConsumerConfig { + set, ok := inputSpecs.(*schema.Set) + if !ok || set.Len() == 0 { + return nil + } + + specs := make(map[string]utils.ConsumerConfig, set.Len()) + for _, item := range set.List() { + spec, ok := item.(map[string]interface{}) + if !ok { + continue + } + + topic, _ := spec[resourceFunctionInputSpecTopicKey].(string) + if topic == "" { + continue + } + + consumerConfig := utils.ConsumerConfig{} + if v, ok := spec[resourceFunctionInputSpecSchemaTypeKey].(string); ok { + consumerConfig.SchemaType = v + } + if v, ok := spec[resourceFunctionInputSpecSerdeClassNameKey].(string); ok { + consumerConfig.SerdeClassName = v + } + if v, ok := spec[resourceFunctionInputSpecRegexPatternKey].(bool); ok { + consumerConfig.RegexPattern = v + } + if v, ok := spec[resourceFunctionInputSpecReceiverQueueSizeKey].(int); ok { + consumerConfig.SetReceiverQueueSize(v) + } + if v, ok := spec[resourceFunctionInputSpecPoolMessagesKey].(bool); ok { + consumerConfig.PoolMessages = v + } + consumerConfig.SchemaProperties = functionStringMap(spec[resourceFunctionInputSpecSchemaPropertiesKey]) + consumerConfig.ConsumerProperties = functionStringMap(spec[resourceFunctionInputSpecConsumerPropertiesKey]) + + specs[topic] = consumerConfig + } + + if len(specs) == 0 { + return nil + } + + return specs +} + +// functionStringMap narrows a schema.TypeMap value to map[string]string, returning nil when empty so +// the field is omitted from the request payload. +func functionStringMap(value interface{}) map[string]string { + interMap, ok := value.(map[string]interface{}) + if !ok || len(interMap) == 0 { + return nil + } + + stringMap := make(map[string]string, len(interMap)) + for key, item := range interMap { + stringMap[key], _ = item.(string) + } + + return stringMap +} + +func functionLegacyInputMap( + value interface{}, inputSpecs map[string]utils.ConsumerConfig, +) map[string]string { + stringMap := functionStringMap(value) + for topic := range inputSpecs { + delete(stringMap, topic) + } + if len(stringMap) == 0 { + return nil + } + + return stringMap +} + func resourcePulsarFunctionRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics { client := getV3ClientFromMeta(meta).Functions() @@ -587,12 +1014,28 @@ func marshalFunctionConfig(d *schema.ResourceData) (*utils.FunctionConfig, error functionConfig.Name = inter.(string) } + inputSpecs := functionInputSpecsFromSchema(d.Get(resourceFunctionInputSpecsKey)) + if len(inputSpecs) != 0 { + functionConfig.InputSpecs = inputSpecs + } + if inter, ok := d.GetOk(resourceFunctionInputsKey); ok { inputsSet := inter.(*schema.Set) var inputs []string for _, item := range inputsSet.List() { - inputs = append(inputs, item.(string)) + topic := item.(string) + + // A topic carried by an input_specs block is fully described there, and listing it in + // both places is actively harmful on update: FunctionConfigUtils.validateUpdate() folds + // inputs into the inputSpecs map with a default ConsumerConfig *before* reading that map + // back, so the topic's consumer settings would be discarded on every apply. (The create + // path applies inputSpecs last and does not have this problem, hence the asymmetry.) + if _, ok := inputSpecs[topic]; ok { + continue + } + + inputs = append(inputs, topic) } functionConfig.Inputs = inputs @@ -604,7 +1047,9 @@ func marshalFunctionConfig(d *schema.ResourceData) (*utils.FunctionConfig, error if inter, ok := d.GetOk(resourceFunctionTopicsPatternKey); ok { pattern := inter.(string) - functionConfig.TopicsPattern = &pattern + if _, isInputSpec := inputSpecs[pattern]; !isInputSpec { + functionConfig.TopicsPattern = &pattern + } } if inter, ok := d.GetOk(resourceFunctionJarKey); ok { @@ -701,25 +1146,11 @@ func marshalFunctionConfig(d *schema.ResourceData) (*utils.FunctionConfig, error } if inter, ok := d.GetOk(resourceFunctionCustomSerdeInputsKey); ok { - interMap := inter.(map[string]interface{}) - stringMap := make(map[string]string, len(interMap)) - - for key, value := range interMap { - stringMap[key] = value.(string) - } - - functionConfig.CustomSerdeInputs = stringMap + functionConfig.CustomSerdeInputs = functionLegacyInputMap(inter, inputSpecs) } if inter, ok := d.GetOk(resourceFunctionCustomSchemaInputsKey); ok { - interMap := inter.(map[string]interface{}) - stringMap := make(map[string]string, len(interMap)) - - for key, value := range interMap { - stringMap[key] = value.(string) - } - - functionConfig.CustomSchemaInputs = stringMap + functionConfig.CustomSchemaInputs = functionLegacyInputMap(inter, inputSpecs) } if inter, ok := d.GetOk(resourceFunctionCustomSchemaOutputsKey); ok { @@ -776,6 +1207,96 @@ func marshalFunctionConfig(d *schema.ResourceData) (*utils.FunctionConfig, error return functionConfig, nil } +// unmarshalFunctionInputSpecs writes the server's inputSpecs into state. +// +// It cannot mirror the response verbatim. FunctionConfigUtils.convertFromDetails() returns an entry +// for *every* input topic the function consumes - including topics declared through all four legacy +// input fields - and never reconstructs those legacy fields. Copying all of it into state would +// invent input_specs blocks for configurations that never wrote one, and those would show as a diff +// on every plan forever. +// +// So a returned spec is skipped when the configuration already represents that topic through a +// legacy field, unless it also declares the topic in input_specs (in which case the block is +// genuinely the user's and must be refreshed rather than dropped). On import every legacy field is +// empty, so every spec lands in input_specs - the complete representation of the function. +func unmarshalFunctionInputSpecs(functionConfig utils.FunctionConfig, d *schema.ResourceData) error { + covered := map[string]bool{} + if inter, ok := d.GetOk(resourceFunctionInputsKey); ok { + for _, item := range inter.(*schema.Set).List() { + covered[item.(string)] = true + } + } + if inter, ok := d.GetOk(resourceFunctionTopicsPatternKey); ok { + covered[inter.(string)] = true + } + for _, key := range []string{ + resourceFunctionCustomSerdeInputsKey, + resourceFunctionCustomSchemaInputsKey, + } { + if inter, ok := d.GetOk(key); ok { + for topic := range functionInputMapKeys(inter) { + covered[topic] = true + } + } + } + + declared := functionInputSpecsFromSchema(d.Get(resourceFunctionInputSpecsKey)) + + specs := make([]interface{}, 0, len(functionConfig.InputSpecs)) + for topic, consumerConfig := range functionConfig.InputSpecs { + declaredConfig, isDeclared := declared[topic] + if covered[topic] && !isDeclared { + continue + } + + // Pulsar 4.0.x persists consumerProperties but convertFromDetails() does not copy them into + // the FunctionConfig returned by GET. Preserve the configured state until the server can + // round-trip the field. A user removing the map has an empty declared value, so removal still + // reaches both the request and state. + if isDeclared && len(consumerConfig.ConsumerProperties) == 0 && + len(declaredConfig.ConsumerProperties) != 0 { + consumerConfig.ConsumerProperties = declaredConfig.ConsumerProperties + } + + specs = append(specs, flattenFunctionInputSpec(topic, consumerConfig)) + } + + return d.Set(resourceFunctionInputSpecsKey, specs) +} + +func flattenFunctionInputSpec(topic string, consumerConfig utils.ConsumerConfig) map[string]interface{} { + spec := map[string]interface{}{ + resourceFunctionInputSpecTopicKey: topic, + resourceFunctionInputSpecReceiverQueueSizeKey: defaultFunctionReceiverQueueSize, + resourceFunctionInputSpecRegexPatternKey: consumerConfig.RegexPattern, + resourceFunctionInputSpecPoolMessagesKey: consumerConfig.PoolMessages, + } + + if consumerConfig.HasReceiverQueueSize() { + spec[resourceFunctionInputSpecReceiverQueueSizeKey] = consumerConfig.ReceiverQueueSize + } + + if consumerConfig.SchemaType != "" { + spec[resourceFunctionInputSpecSchemaTypeKey] = consumerConfig.SchemaType + } + + if consumerConfig.SerdeClassName != "" { + spec[resourceFunctionInputSpecSerdeClassNameKey] = consumerConfig.SerdeClassName + } + + // convertFromDetails always returns these maps non-nil, usually empty. Only surface them when + // they hold something, so an empty map does not read as configuration the user never wrote. + if len(consumerConfig.SchemaProperties) != 0 { + spec[resourceFunctionInputSpecSchemaPropertiesKey] = convertToInterfaceMap(consumerConfig.SchemaProperties) + } + + if len(consumerConfig.ConsumerProperties) != 0 { + spec[resourceFunctionInputSpecConsumerPropertiesKey] = convertToInterfaceMap(consumerConfig.ConsumerProperties) + } + + return spec +} + func unmarshalFunctionConfig(functionConfig utils.FunctionConfig, d *schema.ResourceData) error { if functionConfig.Jar != nil { err := d.Set(resourceFunctionJarKey, *functionConfig.Jar) @@ -822,6 +1343,10 @@ func unmarshalFunctionConfig(functionConfig utils.FunctionConfig, d *schema.Reso } } + if err := unmarshalFunctionInputSpecs(functionConfig, d); err != nil { + return err + } + if functionConfig.Parallelism != 0 { err := d.Set(resourceFunctionParallelismKey, functionConfig.Parallelism) if err != nil { diff --git a/pulsar/resource_pulsar_function_test.go b/pulsar/resource_pulsar_function_test.go index 30071d63..9d0333d8 100644 --- a/pulsar/resource_pulsar_function_test.go +++ b/pulsar/resource_pulsar_function_test.go @@ -41,6 +41,20 @@ func TestFunction(t *testing.T) { t.Fatal(err) } + updatedConfigBytes, err := os.ReadFile("testdata/function/main_updated.tf") + if err != nil { + t.Fatal(err) + } + + zeroQueueConfigBytes, err := os.ReadFile("testdata/function/main_zero_queue.tf") + if err != nil { + t.Fatal(err) + } + + // Captured after create so the update step can prove the function was updated in place rather + // than destroyed and recreated. + var createdID string + resource.Test(t, resource.TestCase{ PreCheck: func() { testAccPreCheck(t) }, ProviderFactories: testAccProviderFactories, @@ -66,6 +80,8 @@ func TestFunction(t *testing.T) { } fmt.Printf("config: %v\n", config) + createdID = rs.Primary.ID + assert.Equal(t, "function-1", config.Name) assert.Equal(t, "public", config.Tenant) assert.Equal(t, "default", config.Namespace) @@ -74,9 +90,136 @@ func TestFunction(t *testing.T) { assert.Equal(t, int64(6666), *config.TimeoutMs) assert.NotNil(t, config.Resources) + // input1 carries an input_specs block; input2 is a plain input. Pulsar returns a + // spec for both regardless. + assert.Equal(t, 100, config.InputSpecs["public/default/input1"].ReceiverQueueSize) + assert.Equal(t, "avro", config.InputSpecs["public/default/input1"].SchemaType) + assert.Contains(t, config.InputSpecs, "public/default/input2") + assert.Equal(t, 101, + config.InputSpecs["public/default/pattern-.*"].ReceiverQueueSize) + assert.True(t, config.InputSpecs["public/default/pattern-.*"].RegexPattern) + assert.Equal(t, 102, + config.InputSpecs["public/default/serde-input"].ReceiverQueueSize) + assert.Equal(t, "org.apache.pulsar.functions.api.utils.DefaultSerDe", + config.InputSpecs["public/default/serde-input"].SerdeClassName) + assert.Equal(t, 103, + config.InputSpecs["public/default/schema-input"].ReceiverQueueSize) + assert.Equal(t, "STRING", config.InputSpecs["public/default/schema-input"].SchemaType) + + return nil + }), + }, + { + // Pulsar accepts a receiver queue size change on an existing topic, so this must be + // an in-place update. + Config: string(updatedConfigBytes), + Check: resource.ComposeTestCheckFunc(func(s *terraform.State) error { + name := "pulsar_function.function-1" + rs, ok := s.RootModule().Resources[name] + if !ok { + return fmt.Errorf("%s not be found", name) + } + + if rs.Primary.ID != createdID { + return fmt.Errorf("function was replaced: id changed from %s to %s", + createdID, rs.Primary.ID) + } + + config, err := getPulsarFunctionByResourceID(rs.Primary.ID) + if err != nil { + return err + } + + if config == nil { + return fmt.Errorf("failed to update %s function", rs.Primary.ID) + } + + assert.Equal(t, 250, config.InputSpecs["public/default/input1"].ReceiverQueueSize) + assert.Equal(t, 251, + config.InputSpecs["public/default/pattern-.*"].ReceiverQueueSize) + assert.Equal(t, 252, + config.InputSpecs["public/default/serde-input"].ReceiverQueueSize) + assert.Equal(t, 253, + config.InputSpecs["public/default/schema-input"].ReceiverQueueSize) + + // The topic listed in both inputs and input_specs must keep its consumer config + // across the update: if the provider left it in inputs, validateUpdate() would + // have reset it to a default ConsumerConfig here. + assert.Equal(t, "avro", config.InputSpecs["public/default/input1"].SchemaType) + return nil }), }, + { + // Zero is a valid explicit value, distinct from an omitted queue size. Verify it + // reaches Pulsar, survives GET, and remains an in-place update. + Config: string(zeroQueueConfigBytes), + Check: resource.ComposeTestCheckFunc(func(s *terraform.State) error { + name := "pulsar_function.function-1" + rs, ok := s.RootModule().Resources[name] + if !ok { + return fmt.Errorf("%s not be found", name) + } + + if rs.Primary.ID != createdID { + return fmt.Errorf("function was replaced: id changed from %s to %s", + createdID, rs.Primary.ID) + } + + config, err := getPulsarFunctionByResourceID(rs.Primary.ID) + if err != nil { + return err + } + if config == nil { + return fmt.Errorf("failed to update %s function", rs.Primary.ID) + } + + consumerConfig := config.InputSpecs["public/default/input1"] + assert.True(t, consumerConfig.HasReceiverQueueSize()) + assert.Zero(t, consumerConfig.ReceiverQueueSize) + + return nil + }), + }, + }, + }) +} + +// Regression guard: a function configured only with `inputs` must not drift. The broker returns an +// inputSpecs entry for every one of those topics, so a read that mirrored them into state would +// plan a change on every run. resource.Test fails the step if the post-apply plan is non-empty. +func TestFunctionLegacyInputsOnly(t *testing.T) { + configBytes, err := os.ReadFile("testdata/function/legacy_inputs_only.tf") + if err != nil { + t.Fatal(err) + } + + resource.Test(t, resource.TestCase{ + PreCheck: func() { testAccPreCheck(t) }, + ProviderFactories: testAccProviderFactories, + PreventPostDestroyRefresh: false, + CheckDestroy: testPulsarFunctionDestroy, + Steps: []resource.TestStep{ + {Config: string(configBytes)}, + }, + }) +} + +// Regression guard for the other legacy input forms. Pulsar returns both as inputSpecs and does not +// reconstruct either custom map on GET, so refresh must not invent input_specs blocks for them. +func TestFunctionLegacyCustomInputs(t *testing.T) { + configBytes, err := os.ReadFile("testdata/function/legacy_custom_inputs.tf") + if err != nil { + t.Fatal(err) + } + + resource.Test(t, resource.TestCase{ + PreCheck: func() { testAccPreCheck(t) }, + ProviderFactories: testAccProviderFactories, + PreventPostDestroyRefresh: false, + CheckDestroy: testPulsarFunctionDestroy, + Steps: []resource.TestStep{ + {Config: string(configBytes)}, }, }) } diff --git a/pulsar/resource_pulsar_function_unit_test.go b/pulsar/resource_pulsar_function_unit_test.go index a56a1f15..d9d1b752 100644 --- a/pulsar/resource_pulsar_function_unit_test.go +++ b/pulsar/resource_pulsar_function_unit_test.go @@ -1,9 +1,14 @@ package pulsar import ( + "context" "encoding/json" + "strings" "testing" + "github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils" + "github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema" + "github.com/hashicorp/terraform-plugin-sdk/v2/terraform" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -64,3 +69,656 @@ func TestSplitFunctionCustomRuntimeOptionsWithoutSinkConfig(t *testing.T) { assert.Nil(t, sourceConfig) assert.JSONEq(t, `{"foo":"bar"}`, sanitized) } + +func functionInputSpec(topic string, overrides map[string]interface{}) map[string]interface{} { + spec := map[string]interface{}{ + resourceFunctionInputSpecTopicKey: topic, + resourceFunctionInputSpecReceiverQueueSizeKey: defaultFunctionReceiverQueueSize, + resourceFunctionInputSpecSchemaTypeKey: "", + resourceFunctionInputSpecSerdeClassNameKey: "", + resourceFunctionInputSpecRegexPatternKey: false, + resourceFunctionInputSpecPoolMessagesKey: false, + resourceFunctionInputSpecSchemaPropertiesKey: map[string]interface{}{}, + resourceFunctionInputSpecConsumerPropertiesKey: map[string]interface{}{}, + } + + for key, value := range overrides { + spec[key] = value + } + + return spec +} + +func functionResourceData(t *testing.T, values map[string]interface{}) *schema.ResourceData { + t.Helper() + + d := schema.TestResourceDataRaw(t, resourcePulsarFunction().Schema, map[string]interface{}{}) + for key, value := range values { + require.NoError(t, d.Set(key, value)) + } + + return d +} + +func functionInputSpecsInState(t *testing.T, d *schema.ResourceData) map[string]map[string]interface{} { + t.Helper() + + set, ok := d.Get(resourceFunctionInputSpecsKey).(*schema.Set) + require.True(t, ok) + + specs := map[string]map[string]interface{}{} + for _, item := range set.List() { + spec := item.(map[string]interface{}) + specs[spec[resourceFunctionInputSpecTopicKey].(string)] = spec + } + + return specs +} + +func TestMarshalFunctionInputSpecs(t *testing.T) { + d := functionResourceData(t, map[string]interface{}{ + resourceFunctionTenantKey: "public", + resourceFunctionNamespaceKey: "default", + resourceFunctionNameKey: "function-1", + resourceFunctionInputsKey: []interface{}{"public/default/in-1", "public/default/in-2"}, + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", map[string]interface{}{ + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + resourceFunctionInputSpecSchemaTypeKey: "avro", + resourceFunctionInputSpecConsumerPropertiesKey: map[string]interface{}{"application": "billing"}, + }), + functionInputSpec("public/default/in-3", map[string]interface{}{ + resourceFunctionInputSpecReceiverQueueSizeKey: 500, + resourceFunctionInputSpecPoolMessagesKey: true, + }), + }, + }) + + functionConfig, err := marshalFunctionConfig(d) + require.NoError(t, err) + + require.Len(t, functionConfig.InputSpecs, 2) + assert.Equal(t, 100, functionConfig.InputSpecs["public/default/in-1"].ReceiverQueueSize) + assert.Equal(t, "avro", functionConfig.InputSpecs["public/default/in-1"].SchemaType) + assert.Equal(t, map[string]string{"application": "billing"}, + functionConfig.InputSpecs["public/default/in-1"].ConsumerProperties) + assert.Equal(t, 500, functionConfig.InputSpecs["public/default/in-3"].ReceiverQueueSize) + assert.True(t, functionConfig.InputSpecs["public/default/in-3"].PoolMessages) + + // Empty maps are dropped rather than sent as {}. + assert.Nil(t, functionConfig.InputSpecs["public/default/in-1"].SchemaProperties) + + // in-1 is declared in input_specs, so it must not also be sent in inputs: validateUpdate() + // would fold it back in with a default ConsumerConfig and discard the receiver queue size. + assert.Equal(t, []string{"public/default/in-2"}, functionConfig.Inputs) +} + +func TestMarshalFunctionInputSpecsAbsent(t *testing.T) { + d := functionResourceData(t, map[string]interface{}{ + resourceFunctionTenantKey: "public", + resourceFunctionNamespaceKey: "default", + resourceFunctionNameKey: "function-1", + resourceFunctionInputsKey: []interface{}{"public/default/in-1"}, + }) + + functionConfig, err := marshalFunctionConfig(d) + require.NoError(t, err) + + assert.Nil(t, functionConfig.InputSpecs) + assert.Equal(t, []string{"public/default/in-1"}, functionConfig.Inputs) +} + +func TestMarshalFunctionInputSpecsExplicitZeroQueueSize(t *testing.T) { + d := functionResourceData(t, map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", map[string]interface{}{ + resourceFunctionInputSpecReceiverQueueSizeKey: 0, + }), + }, + }) + + functionConfig, err := marshalFunctionConfig(d) + require.NoError(t, err) + consumerConfig := functionConfig.InputSpecs["public/default/in-1"] + assert.True(t, consumerConfig.HasReceiverQueueSize()) + assert.Zero(t, consumerConfig.ReceiverQueueSize) + + payload, err := json.Marshal(functionConfig) + require.NoError(t, err) + assert.Contains(t, string(payload), `"receiverQueueSize":0`) +} + +func TestMarshalFunctionInputSpecsRemoveEveryLegacyOverlap(t *testing.T) { + const defaultSerde = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + + d := functionResourceData(t, map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{ + "public/default/plain", + "public/default/keep-input", + }, + resourceFunctionTopicsPatternKey: "public/default/pattern-.*", + resourceFunctionCustomSerdeInputsKey: map[string]interface{}{ + "public/default/serde": defaultSerde, + "public/default/keep-serde": defaultSerde, + }, + resourceFunctionCustomSchemaInputsKey: map[string]interface{}{ + "public/default/schema": `{"schemaType":"STRING"}`, + "public/default/keep-schema": `{"schemaType":"STRING"}`, + }, + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/plain", nil), + functionInputSpec("public/default/pattern-.*", map[string]interface{}{ + resourceFunctionInputSpecRegexPatternKey: true, + }), + functionInputSpec("public/default/serde", map[string]interface{}{ + resourceFunctionInputSpecSerdeClassNameKey: defaultSerde, + }), + functionInputSpec("public/default/schema", map[string]interface{}{ + resourceFunctionInputSpecSchemaTypeKey: "STRING", + }), + }, + }) + + functionConfig, err := marshalFunctionConfig(d) + require.NoError(t, err) + + assert.Equal(t, []string{"public/default/keep-input"}, functionConfig.Inputs) + assert.Nil(t, functionConfig.TopicsPattern) + assert.Equal(t, map[string]string{ + "public/default/keep-serde": defaultSerde, + }, functionConfig.CustomSerdeInputs) + assert.Equal(t, map[string]string{ + "public/default/keep-schema": `{"schemaType":"STRING"}`, + }, functionConfig.CustomSchemaInputs) +} + +func TestUnmarshalFunctionInputSpecs(t *testing.T) { + // The broker returns a spec for every input topic, including ones the configuration declares + // through inputs or topics_pattern, and never returns inputs at all. + functionConfig := utils.FunctionConfig{ + InputSpecs: map[string]utils.ConsumerConfig{ + "public/default/in-1": { + ReceiverQueueSize: 100, + SchemaType: "avro", + SchemaProperties: map[string]string{}, + ConsumerProperties: map[string]string{"application": "billing"}, + }, + "public/default/in-2": { + SchemaProperties: map[string]string{}, + ConsumerProperties: map[string]string{}, + }, + "public/default/pattern-.*": { + RegexPattern: true, + SchemaProperties: map[string]string{}, + ConsumerProperties: map[string]string{}, + }, + "public/default/serde": {}, + "public/default/schema": { + SchemaType: "STRING", + }, + }, + } + + d := functionResourceData(t, map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{"public/default/in-1", "public/default/in-2"}, + resourceFunctionTopicsPatternKey: "public/default/pattern-.*", + resourceFunctionCustomSerdeInputsKey: map[string]interface{}{ + "public/default/serde": "org.apache.pulsar.functions.api.utils.DefaultSerDe", + }, + resourceFunctionCustomSchemaInputsKey: map[string]interface{}{ + "public/default/schema": `{"schemaType":"STRING"}`, + }, + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", map[string]interface{}{ + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }), + }, + }) + + require.NoError(t, unmarshalFunctionInputSpecs(functionConfig, d)) + + specs := functionInputSpecsInState(t, d) + + // in-1 is declared in input_specs, so it is refreshed even though inputs also lists it. + require.Contains(t, specs, "public/default/in-1") + assert.Equal(t, 100, specs["public/default/in-1"][resourceFunctionInputSpecReceiverQueueSizeKey]) + assert.Equal(t, "avro", specs["public/default/in-1"][resourceFunctionInputSpecSchemaTypeKey]) + assert.Equal(t, map[string]interface{}{"application": "billing"}, + specs["public/default/in-1"][resourceFunctionInputSpecConsumerPropertiesKey]) + + // in-2 is represented by inputs and the pattern by topics_pattern; surfacing either would be a + // block the configuration never wrote, and therefore a permanent diff. + assert.NotContains(t, specs, "public/default/in-2") + assert.NotContains(t, specs, "public/default/pattern-.*") + assert.NotContains(t, specs, "public/default/serde") + assert.NotContains(t, specs, "public/default/schema") +} + +func TestUnmarshalFunctionInputSpecsPreservesConsumerProperties(t *testing.T) { + topic := "public/default/in-1" + d := functionResourceData(t, map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec(topic, map[string]interface{}{ + resourceFunctionInputSpecConsumerPropertiesKey: map[string]interface{}{ + "application": "billing", + }, + }), + }, + }) + + functionConfig := utils.FunctionConfig{ + InputSpecs: map[string]utils.ConsumerConfig{ + topic: { + ReceiverQueueSize: defaultFunctionReceiverQueueSize, + ConsumerProperties: map[string]string{}, + }, + }, + } + + require.NoError(t, unmarshalFunctionInputSpecs(functionConfig, d)) + + specs := functionInputSpecsInState(t, d) + assert.Equal(t, map[string]interface{}{"application": "billing"}, + specs[topic][resourceFunctionInputSpecConsumerPropertiesKey]) +} + +func TestUnmarshalFunctionInputSpecsOnImport(t *testing.T) { + // On import nothing is in state yet, so every spec the broker returns is the function's + // complete input configuration. + functionConfig := utils.FunctionConfig{ + InputSpecs: map[string]utils.ConsumerConfig{ + "public/default/in-1": {ReceiverQueueSize: 100}, + "public/default/in-2": {}, + }, + } + + d := functionResourceData(t, map[string]interface{}{}) + require.NoError(t, unmarshalFunctionInputSpecs(functionConfig, d)) + + specs := functionInputSpecsInState(t, d) + assert.Len(t, specs, 2) + assert.Equal(t, 100, specs["public/default/in-1"][resourceFunctionInputSpecReceiverQueueSizeKey]) + assert.Equal(t, defaultFunctionReceiverQueueSize, + specs["public/default/in-2"][resourceFunctionInputSpecReceiverQueueSizeKey]) +} + +func TestUnmarshalFunctionInputSpecsExplicitZeroQueueSize(t *testing.T) { + consumerConfig := utils.ConsumerConfig{} + consumerConfig.SetReceiverQueueSize(0) + + d := functionResourceData(t, map[string]interface{}{}) + require.NoError(t, unmarshalFunctionInputSpecs(utils.FunctionConfig{ + InputSpecs: map[string]utils.ConsumerConfig{ + "public/default/in-1": consumerConfig, + }, + }, d)) + + specs := functionInputSpecsInState(t, d) + assert.Zero(t, specs["public/default/in-1"][resourceFunctionInputSpecReceiverQueueSizeKey]) +} + +func TestEffectiveFunctionInputTopics(t *testing.T) { + d := functionResourceData(t, map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{"public/default/in-1", "public/default/in-2"}, + resourceFunctionTopicsPatternKey: "public/default/pattern-.*", + resourceFunctionCustomSerdeInputsKey: map[string]interface{}{ + "public/default/serde": "org.apache.pulsar.functions.api.utils.DefaultSerDe", + }, + resourceFunctionCustomSchemaInputsKey: map[string]interface{}{ + "public/default/schema": `{"schemaType":"STRING"}`, + }, + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", map[string]interface{}{ + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }), + functionInputSpec("public/default/in-3", nil), + }, + }) + + topics := effectiveFunctionInputTopics( + d.Get(resourceFunctionInputsKey), + d.Get(resourceFunctionTopicsPatternKey), + d.Get(resourceFunctionCustomSerdeInputsKey), + d.Get(resourceFunctionCustomSchemaInputsKey), + d.Get(resourceFunctionInputSpecsKey), + ) + + assert.Equal(t, map[string]bool{ + "public/default/in-1": false, + "public/default/in-2": false, + "public/default/in-3": false, + "public/default/pattern-.*": true, + "public/default/serde": false, + "public/default/schema": false, + }, topics) +} + +func TestFunctionInputSpecsValidation(t *testing.T) { + base := map[string]interface{}{ + resourceFunctionTenantKey: "public", + resourceFunctionNamespaceKey: "default", + resourceFunctionNameKey: "function-1", + } + + tests := []struct { + name string + specs []interface{} + want string + }{ + { + name: "duplicate topic keys", + specs: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }, + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecReceiverQueueSizeKey: 200, + }, + }, + want: `input_specs contains duplicate key "public/default/in-1"`, + }, + { + name: "schema and serde are mutually exclusive", + specs: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecSchemaTypeKey: "STRING", + resourceFunctionInputSpecSerdeClassNameKey: "example.StringSerde", + }, + }, + want: "cannot set both schema_type and serde_class_name", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + config := map[string]interface{}{} + for key, value := range base { + config[key] = value + } + config[resourceFunctionInputSpecsKey] = test.specs + + res := resourcePulsarFunction() + d := schema.TestResourceDataRaw(t, res.Schema, map[string]interface{}{}) + _, err := res.Diff( + context.Background(), + d.State(), + terraform.NewResourceConfigRaw(config), + nil, + ) + require.ErrorContains(t, err, test.want) + }) + } +} + +func functionInputSpecsDiff(t *testing.T, state, config map[string]interface{}) *terraform.InstanceDiff { + t.Helper() + + res := resourcePulsarFunction() + + d := schema.TestResourceDataRaw(t, res.Schema, map[string]interface{}{}) + for key, value := range state { + require.NoError(t, d.Set(key, value)) + } + d.SetId("public/default/function-1") + + diff, err := res.Diff(context.Background(), d.State(), terraform.NewResourceConfigRaw(config), nil) + require.NoError(t, err) + require.NotNil(t, diff) + + return diff +} + +// inputSpecsDiffContains reports whether any input_specs attribute is planned to become want. The +// in-place cases assert on this so that a change the SDK dropped entirely - diffSet returns early +// when the element hashes match - cannot pass as "no replacement needed". +func inputSpecsDiffContains(diff *terraform.InstanceDiff, want string) bool { + for key, attr := range diff.Attributes { + if strings.HasPrefix(key, resourceFunctionInputSpecsKey+".") && attr != nil && attr.New == want { + return true + } + } + + return false +} + +// Pulsar accepts consumer-setting changes in place but rejects changes to the set of input topics, +// so the provider must replace the function in exactly the second case and no other. +func TestFunctionInputSpecsForceNew(t *testing.T) { + base := map[string]interface{}{ + resourceFunctionTenantKey: "public", + resourceFunctionNamespaceKey: "default", + resourceFunctionNameKey: "function-1", + } + + withBase := func(values map[string]interface{}) map[string]interface{} { + merged := map[string]interface{}{} + for key, value := range base { + merged[key] = value + } + for key, value := range values { + merged[key] = value + } + return merged + } + + tests := []struct { + name string + state map[string]interface{} + config map[string]interface{} + requiresNew bool + // planned value that must appear somewhere in the input_specs diff + plans string + }{ + { + name: "adopting input_specs for a topic already in inputs updates in place", + state: withBase(map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{"public/default/in-1", "public/default/in-2"}, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{"public/default/in-1", "public/default/in-2"}, + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }, + }, + }), + requiresNew: false, + plans: "100", + }, + { + name: "moving a topic from inputs to input_specs updates in place", + state: withBase(map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{"public/default/in-1"}, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }, + }, + }), + requiresNew: false, + plans: "100", + }, + { + name: "moving a pattern to input_specs updates in place", + state: withBase(map[string]interface{}{ + resourceFunctionTopicsPatternKey: "public/default/in-.*", + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-.*", + resourceFunctionInputSpecRegexPatternKey: true, + }, + }, + }), + requiresNew: false, + }, + { + name: "moving custom serde input to input_specs updates in place", + state: withBase(map[string]interface{}{ + resourceFunctionCustomSerdeInputsKey: map[string]interface{}{ + "public/default/in-1": "org.apache.pulsar.functions.api.utils.DefaultSerDe", + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecSerdeClassNameKey: "org.apache.pulsar.functions.api.utils." + + "DefaultSerDe", + }, + }, + }), + requiresNew: false, + }, + { + name: "changing a custom serde for the same topic updates in place", + state: withBase(map[string]interface{}{ + resourceFunctionCustomSerdeInputsKey: map[string]interface{}{ + "public/default/in-1": "example.OldSerde", + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionCustomSerdeInputsKey: map[string]interface{}{ + "public/default/in-1": "example.NewSerde", + }, + }), + requiresNew: false, + }, + { + name: "tuning receiver_queue_size updates in place", + state: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", map[string]interface{}{ + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }), + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecReceiverQueueSizeKey: 250, + }, + }, + }), + requiresNew: false, + plans: "250", + }, + { + name: "adding a new input topic replaces the function", + state: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", map[string]interface{}{ + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }), + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + resourceFunctionInputSpecReceiverQueueSizeKey: 100, + }, + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-2", + }, + }, + }), + requiresNew: true, + }, + { + name: "renaming an input topic replaces the function", + state: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", nil), + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/renamed", + }, + }, + }), + requiresNew: true, + }, + { + name: "flipping regex_pattern replaces the function", + state: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-.*", nil), + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-.*", + resourceFunctionInputSpecRegexPatternKey: true, + }, + }, + }), + requiresNew: true, + }, + { + name: "dropping an input topic replaces the function", + state: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + functionInputSpec("public/default/in-1", nil), + functionInputSpec("public/default/in-2", nil), + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputSpecsKey: []interface{}{ + map[string]interface{}{ + resourceFunctionInputSpecTopicKey: "public/default/in-1", + }, + }, + }), + requiresNew: true, + }, + { + name: "renaming a plain input replaces the function", + state: withBase(map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{"public/default/in-1"}, + }), + config: withBase(map[string]interface{}{ + resourceFunctionInputsKey: []interface{}{"public/default/in-2"}, + }), + requiresNew: true, + }, + { + name: "renaming a custom schema input replaces the function", + state: withBase(map[string]interface{}{ + resourceFunctionCustomSchemaInputsKey: map[string]interface{}{ + "public/default/in-1": `{"schemaType":"STRING"}`, + }, + }), + config: withBase(map[string]interface{}{ + resourceFunctionCustomSchemaInputsKey: map[string]interface{}{ + "public/default/in-2": `{"schemaType":"STRING"}`, + }, + }), + requiresNew: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + diff := functionInputSpecsDiff(t, test.state, test.config) + assert.Equal(t, test.requiresNew, diff.RequiresNew()) + if test.plans != "" { + assert.True(t, inputSpecsDiffContains(diff, test.plans), + "expected the diff to plan %q somewhere under %s, got %v", + test.plans, resourceFunctionInputSpecsKey, diff.Attributes) + } + }) + } +} diff --git a/pulsar/testdata/function/legacy_custom_inputs.tf b/pulsar/testdata/function/legacy_custom_inputs.tf new file mode 100644 index 00000000..f451e802 --- /dev/null +++ b/pulsar/testdata/function/legacy_custom_inputs.tf @@ -0,0 +1,28 @@ + +provider "pulsar" { + web_service_url = "http://localhost:8080" +} + +resource "pulsar_function" "function-legacy-custom-inputs" { + provider = pulsar + + name = "function-legacy-custom-inputs" + tenant = "public" + namespace = "default" + parallelism = 1 + + processing_guarantees = "ATLEAST_ONCE" + + jar = "function://public/default/api-examples@v1" + classname = "org.apache.pulsar.functions.api.examples.WordCountFunction" + + custom_serde_inputs = { + "public/default/legacy-serde-input" = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + } + + custom_schema_inputs = { + "public/default/legacy-schema-input" = jsonencode({ schemaType = "STRING" }) + } + + output = "public/default/legacy-custom-output" +} diff --git a/pulsar/testdata/function/legacy_inputs_only.tf b/pulsar/testdata/function/legacy_inputs_only.tf new file mode 100644 index 00000000..83e79317 --- /dev/null +++ b/pulsar/testdata/function/legacy_inputs_only.tf @@ -0,0 +1,41 @@ + +provider "pulsar" { + web_service_url = "http://localhost:8080" +} + +resource "pulsar_function" "function-1" { + provider = pulsar + + name = "function-1" + tenant = "public" + namespace = "default" + parallelism = 1 + + processing_guarantees = "ATLEAST_ONCE" + + jar = "function://public/default/api-examples@v1" + classname = "org.apache.pulsar.functions.api.examples.WordCountFunction" + + inputs = ["public/default/input1", "public/default/input2"] + + + output = "public/default/test-out" + + subscription_name = "tf-sub" + subscription_position = "Latest" + cleanup_subscription = true + forward_source_message_property = true + retain_key_ordering = true + auto_ack = true + max_message_retries = 101 + dead_letter_topic = "public/default/dlt" + log_topic = "public/default/lt" + timeout_ms = 6666 + + custom_runtime_options = jsonencode( + { + "env": { + "PULSAR": "FUNCTIONS" + } + }) +} \ No newline at end of file diff --git a/pulsar/testdata/function/main.tf b/pulsar/testdata/function/main.tf index 8997df4f..0f68db35 100644 --- a/pulsar/testdata/function/main.tf +++ b/pulsar/testdata/function/main.tf @@ -4,37 +4,79 @@ provider "pulsar" { } resource "pulsar_function" "function-1" { - provider = pulsar + provider = pulsar - name = "function-1" - tenant = "public" - namespace = "default" - parallelism = 1 + name = "function-1" + tenant = "public" + namespace = "default" + parallelism = 1 - processing_guarantees = "ATLEAST_ONCE" + processing_guarantees = "ATLEAST_ONCE" - jar = "function://public/default/api-examples@v1" - classname = "org.apache.pulsar.functions.api.examples.WordCountFunction" + jar = "function://public/default/api-examples@v1" + classname = "org.apache.pulsar.functions.api.examples.WordCountFunction" - inputs = ["public/default/input1", "public/default/input2"] + inputs = ["public/default/input1", "public/default/input2"] - output = "public/default/test-out" + topics_pattern = "public/default/pattern-.*" - subscription_name = "tf-sub" - subscription_position = "Latest" - cleanup_subscription = true - forward_source_message_property = true - retain_key_ordering = true - auto_ack = true - max_message_retries = 101 - dead_letter_topic = "public/default/dlt" - log_topic = "public/default/lt" - timeout_ms = 6666 + custom_serde_inputs = { + "public/default/serde-input" = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + } - custom_runtime_options = jsonencode( + custom_schema_inputs = { + "public/default/schema-input" = jsonencode({ schemaType = "STRING" }) + } + + # input1 is deliberately listed in both inputs and input_specs: the provider must strip it from + # inputs on the wire, or Pulsar's validateUpdate() folds it back in with a default + # ConsumerConfig and discards the receiver queue size on every apply. + input_specs { + key = "public/default/input1" + receiver_queue_size = 100 + schema_type = "avro" + consumer_properties = { + application = "billing" + } + } + + # Each legacy input form is deliberately overlapped. input_specs must win on both create and + # update, even though Pulsar's update path otherwise applies the legacy forms last. + input_specs { + key = "public/default/pattern-.*" + receiver_queue_size = 101 + is_regex_pattern = true + } + + input_specs { + key = "public/default/serde-input" + receiver_queue_size = 102 + serde_class_name = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + } + + input_specs { + key = "public/default/schema-input" + receiver_queue_size = 103 + schema_type = "STRING" + } + + output = "public/default/test-out" + + subscription_name = "tf-sub" + subscription_position = "Latest" + cleanup_subscription = true + forward_source_message_property = true + retain_key_ordering = true + auto_ack = true + max_message_retries = 101 + dead_letter_topic = "public/default/dlt" + log_topic = "public/default/lt" + timeout_ms = 6666 + + custom_runtime_options = jsonencode( { - "env": { - "PULSAR": "FUNCTIONS" - } - }) -} \ No newline at end of file + "env" : { + "PULSAR" : "FUNCTIONS" + } + }) +} diff --git a/pulsar/testdata/function/main_updated.tf b/pulsar/testdata/function/main_updated.tf new file mode 100644 index 00000000..710e174b --- /dev/null +++ b/pulsar/testdata/function/main_updated.tf @@ -0,0 +1,80 @@ + +provider "pulsar" { + web_service_url = "http://localhost:8080" +} + +resource "pulsar_function" "function-1" { + provider = pulsar + + name = "function-1" + tenant = "public" + namespace = "default" + parallelism = 1 + + processing_guarantees = "ATLEAST_ONCE" + + jar = "function://public/default/api-examples@v1" + classname = "org.apache.pulsar.functions.api.examples.WordCountFunction" + + inputs = ["public/default/input1", "public/default/input2"] + + topics_pattern = "public/default/pattern-.*" + + custom_serde_inputs = { + "public/default/serde-input" = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + } + + custom_schema_inputs = { + "public/default/schema-input" = jsonencode({ schemaType = "STRING" }) + } + + # input1 is deliberately listed in both inputs and input_specs: the provider must strip it from + # inputs on the wire, or Pulsar's validateUpdate() folds it back in with a default + # ConsumerConfig and discards the receiver queue size on every apply. + input_specs { + key = "public/default/input1" + receiver_queue_size = 250 + schema_type = "avro" + consumer_properties = { + application = "billing" + } + } + + input_specs { + key = "public/default/pattern-.*" + receiver_queue_size = 251 + is_regex_pattern = true + } + + input_specs { + key = "public/default/serde-input" + receiver_queue_size = 252 + serde_class_name = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + } + + input_specs { + key = "public/default/schema-input" + receiver_queue_size = 253 + schema_type = "STRING" + } + + output = "public/default/test-out" + + subscription_name = "tf-sub" + subscription_position = "Latest" + cleanup_subscription = true + forward_source_message_property = true + retain_key_ordering = true + auto_ack = true + max_message_retries = 101 + dead_letter_topic = "public/default/dlt" + log_topic = "public/default/lt" + timeout_ms = 6666 + + custom_runtime_options = jsonencode( + { + "env" : { + "PULSAR" : "FUNCTIONS" + } + }) +} diff --git a/pulsar/testdata/function/main_zero_queue.tf b/pulsar/testdata/function/main_zero_queue.tf new file mode 100644 index 00000000..e9580161 --- /dev/null +++ b/pulsar/testdata/function/main_zero_queue.tf @@ -0,0 +1,77 @@ + +provider "pulsar" { + web_service_url = "http://localhost:8080" +} + +resource "pulsar_function" "function-1" { + provider = pulsar + + name = "function-1" + tenant = "public" + namespace = "default" + parallelism = 1 + + processing_guarantees = "ATLEAST_ONCE" + + jar = "function://public/default/api-examples@v1" + classname = "org.apache.pulsar.functions.api.examples.WordCountFunction" + + inputs = ["public/default/input1", "public/default/input2"] + + topics_pattern = "public/default/pattern-.*" + + custom_serde_inputs = { + "public/default/serde-input" = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + } + + custom_schema_inputs = { + "public/default/schema-input" = jsonencode({ schemaType = "STRING" }) + } + + input_specs { + key = "public/default/input1" + receiver_queue_size = 0 + schema_type = "avro" + consumer_properties = { + application = "billing" + } + } + + input_specs { + key = "public/default/pattern-.*" + receiver_queue_size = 251 + is_regex_pattern = true + } + + input_specs { + key = "public/default/serde-input" + receiver_queue_size = 252 + serde_class_name = "org.apache.pulsar.functions.api.utils.DefaultSerDe" + } + + input_specs { + key = "public/default/schema-input" + receiver_queue_size = 253 + schema_type = "STRING" + } + + output = "public/default/test-out" + + subscription_name = "tf-sub" + subscription_position = "Latest" + cleanup_subscription = true + forward_source_message_property = true + retain_key_ordering = true + auto_ack = true + max_message_retries = 101 + dead_letter_topic = "public/default/dlt" + log_topic = "public/default/lt" + timeout_ms = 6666 + + custom_runtime_options = jsonencode( + { + "env" : { + "PULSAR" : "FUNCTIONS" + } + }) +}