A Practical Guide to Dapr Core Features: Pub/Sub, Resource Bindings, Actors, Observability, Secrets, and Configuration

This technical tutorial provides a comprehensive, hands-on guide to implementing core Dapr features using Java applications deployed on Kubernetes, extending previous discussions on service invocation and state management.

Publish and Subscribe: The article details configuring a Redis-based pub/sub component via YAML, initializing it with kubectl, and publishing messages using curl. It also covers subscription routing to a Java controller endpoint and verifying message consumption through application logs.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: redis-pubsub
  namespace: default
spec:
  type: pubsub.redis
  version: v1
  metadata:
  - name: redisHost
    value: xx.xx.xx.xx:6379
  - name: redisPassword
    value:

Execute component initialization and start the project:

kubectl apply -f pub.yaml
kubectl create -f http-demo.yaml

Test publishing and verify via Redis:

curl -X POST http://xx.xx.xx.xx:3500/v1.0/publish/redis-pubsub/deathStarStatus -H "Content-Type: application/json"  -d '{
       "status": "completed-test"
     }'
xread streams deathStarStatus 0

Configure subscription and Java controller:

apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:
  name: redis-sub
spec:
  topic: deathStarStatus
  route: /dapr-sample-topic
  pubsubname: redis-pubsub

@PostMapping("/dapr-sample-topic")
public Response<String> sampleTopic() {
  System.out.println("sample-topic");
  return Response.ok("sample-topic");
}

Initialize and verify subscription logs:

kubectl apply -f sub.yaml
kubectl create -f http-demo-dapr.yaml
kubectl logs -f daprnodeapp-xx-xx -c dapr-http-demo

Resource Bindings: Using Kafka as an example, the guide shows how to deploy a Kafka cluster via Helm, configure Dapr input and output bindings, and test bidirectional communication. Output bindings trigger external resources, while input bindings allow applications to react to external events.

helm repo add bitnami https://charts.bitnami.com/bitnami
helm repo update
kubectl create ns kafka
helm install dapr-kafka bitnami/kafka --wait --namespace kafka -f ./kafka-non-persistence.yaml

Kafka configuration and Dapr component setup:

# kafka-non-persistence.yaml
replicas: 1
persistence:
  enabled: false
zookeeper:
  persistence:
    enabled: false
  affinity:
    nodeAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
        nodeSelectorTerms:
          - matchExpressions:
            - key: kubernetes.io/os
              operator: In
              values:
              - linux
            - key: kubernetes.io/arch
              operator: In
              values:
              - amd64
autoCreateTopicsEnable: true
affinity:
  nodeAffinity:
    requiredDuringSchedulingIgnoredDuringExecution:
      nodeSelectorTerms:
        - matchExpressions:
          - key: kubernetes.io/os
            operator: In
            values:
            - linux
          - key: kubernetes.io/arch
            operator: In
            values:
            - amd64

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: sample-topic
spec:
  type: bindings.kafka
  version: v1
  metadata:
  - name: brokers
    value: "xx.xx.xx.xx:9092,xx.xx.xx.xx:9093"
  - name: topics
    value: sample
  - name: consumerGroup
    value: group1
  - name: publishTopic
    value: sample
  - name: authRequired
    value: "false"

Test output and input bindings:

kubectl create -f  http-demo-dapr.yaml
curl -X POST -H 'Content-Type: application/json' http://xx.xx.xx.xx:3500/v1.0/bindings/sample-topic -d '{ "data": { "message": "Hi!" }, "operation": "create" }'
kubectl create -f  http-demo.yaml
kubectl logs -f nodeapp-xx-xx -c http-demo

Actors: Dapr's actor model supports stateless timers and stateful reminders. The article details creating a Redis state store for actors, implementing Java interfaces and classes for actor logic, and managing timer/reminder lifecycles via REST APIs.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: statestore
  namespace: default
spec:
  type: state.redis
  version: v1
  metadata:
  - name: redisHost
    value: xx.xx.xx.xx:6379
  - name: redisPassword
    value: ""
  - name: actorStateStore
    value: "true"

Initialize projects and implement actor logic:

kubectl create -f http-demo.yaml
kubectl create -f http-demo-dapr.yaml

@GetMapping("/dapr-sample-actor/{actorId}")
public Response<Integer> sampleActor(@PathVariable("actorId") String actorId) throws InterruptedException {
    System.out.println("sample-actor");
    int messageNumber = doActor(actorId);
    return Response.ok(messageNumber);
}

@GetMapping("/dapr-sample-actor/clock/{actorId}")
public Response<Integer> sampleActorClock(@PathVariable("actorId") String actorId) {
    System.out.println("sample-actor");
    int messageNumber = doActorClock(actorId);
    return Response.ok(messageNumber);
}

private int doActorClock(String actorId){
    ActorClient client = new ActorClient();
    ActorProxyBuilder<DemoActor> builder = new ActorProxyBuilder(DemoActor.class, client);
    DemoActor actor = builder.build(new ActorId(actorId));
    actor.registerActorTimer();
    return sayMessageToActor(actorId, actor);
}

private int doActor(String actorId) {
    ActorClient client = new ActorClient();
    ActorProxyBuilder<DemoActor> builder = new ActorProxyBuilder(DemoActor.class, client);
    DemoActor actor = builder.build(new ActorId(actorId));
    actor.registerReminder();
    return sayMessageToActor(actorId, actor);
}

private int sayMessageToActor(String actorId, DemoActor actor) {
    int messageNumber = actor.incrementAndGet(1).block();
    String message = String.format("Actor %s said message #%d", actorId, messageNumber);
    actor.say(message);
    return messageNumber;
}

Actor interface and implementation:

@ActorType(name = "DemoActor")
public interface DemoActor {
    void registerReminder();
    @ActorMethod(name = "echo_message")
    String say(String something);
    void clock(String message);
    void registerActorTimer();
    @ActorMethod(returns = Integer.class)
    Mono<Integer> incrementAndGet(int delta);
}

package com.test.dapr.sample.interfaces.http;
import io.dapr.actors.ActorId;
import io.dapr.actors.runtime.AbstractActor;
import io.dapr.actors.runtime.ActorRuntimeContext;
import io.dapr.actors.runtime.Remindable;
import io.dapr.utils.TypeRef;
import reactor.core.publisher.Mono;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Calendar;
import java.util.TimeZone;
public class DemoActorImpl extends AbstractActor implements DemoActor, Remindable<Integer> {
    private static final DateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    public DemoActorImpl(ActorRuntimeContext runtimeContext, ActorId id) {
        super(runtimeContext, id);
    }
    @Override
    public void registerActorTimer() {
        super.registerActorTimer("my-actorTimer", "clock", "ping!", Duration.ofSeconds(5), Duration.ofSeconds(2)).block();
    }
    @Override
    public void registerReminder() {
        super.registerReminder("my-reminder", (int) (Integer.MAX_VALUE * Math.random()), Duration.ofSeconds(10), Duration.ZERO).block();
    }
    @Override
    public String say(String something) {
        Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
        String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
        System.out.println("Server say method for actor " + super.getId() + ": " + (something == null ? "" : something + " @ " + utcNowAsString));
        super.getActorStateManager().set("lastMessage", something).block();
        return utcNowAsString;
    }
    @Override
    public Mono<Integer> incrementAndGet(int delta) {
        return super.getActorStateManager().contains("counter")
                .flatMap(exists -> exists ? super.getActorStateManager().get("counter", int.class) : Mono.just(0))
                .map(c -> c + delta)
                .flatMap(c -> super.getActorStateManager().set("counter", c).thenReturn(c));
    }
    @Override
    public void clock(String message) {
        Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
        String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
        System.out.println("clock for actor " + super.getId() + ": " + (message == null ? "" : message + " @ " + utcNowAsString));
    }
    @Override
    public TypeRef<Integer> getStateType() {
        return TypeRef.INT;
    }
    @Override
    public Mono<Void> receiveReminder(String reminderName, Integer state, Duration dueTime, Duration period) {
        return Mono.fromRunnable(() -> {
            Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
            String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
            String message = String.format("Server reminded actor %s of: %s for %d @ %s", this.getId(), reminderName, state, utcNowAsString);
            System.out.println(message);
        });
    }
}

Test timer and reminder APIs:

curl http://xx.xx.xx.xx:3500/v1.0/invoke/daprnodeapp/method/dapr-sample-actor/clock/88
curl -X DELETE http://xx.xx.xx.xx:3500/v1.0/actors/DemoActor/88/timers/my-actorTimer
curl -X GET http://xx.xx.xx.xx:3500/v1.0/invoke/daprnodeapp/method/dapr-sample-actor/90
curl -X PUT -H "Content-Type: application/json" http://xx.xx.xx.xx:3500/v1.0/actors/DemoActor/90/reminders/echo_message -d '{"dueTime":"0h0m9s0ms","period":"0h0m3s0ms"}'
curl -X DELETE http://xx.xx.xx.xx:3500/v1.0/actors/DemoActor/90/reminders/echo_message

Observability: Covering tracing, logging, and metrics, the tutorial focuses on integrating Zipkin for distributed tracing. It includes deploying Zipkin, configuring Dapr tracing settings, modifying application YAML annotations, and verifying trace propagation through HTTP headers.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: zipkin
  labels:
    app: zipkin
spec:
  replicas: 1
  selector:
    matchLabels:
      app: zipkin
  template:
    metadata:
      labels:
        app: zipkin
    spec:
      containers:
      - name: zipkin
        image: openzipkin/zipkin
        ports:
        - containerPort: 9411
---
kind: Service
apiVersion: v1
metadata:
  name: zipkin
  labels:
    app: zipkin
spec:
  selector:
    app: zipkin
  ports:
  - protocol: TCP
    port: 9411
    targetPort: 9411
  type: ClusterIP

Initialize Zipkin and configure Dapr tracing:

kubectl create -f zipkin.yaml

apiVersion: dapr.io/v1alpha1
kind: Configuration
metadata:
  name: zipkin-config
spec:
  tracing:
    samplingRate: "1"
    zipkin:
      endpointAddress: "http://xx.xx.xx.xx:9411/api/v2/spans"

Update project YAML and controller:

dapr.io/log-as-json: "true"
dapr.io/config: "zipkin-config"

@GetMapping("/dapr-helloWorld")
public Response<String> getHelloWorld(@RequestHeader Map<String, String> headers) {
  System.out.println("getHelloWorld");
  HttpExtension httpExtension = new HttpExtension(DaprHttp.HttpMethods.GET, null, headers);
  InvokeMethodRequest request = new InvokeMethodRequest("nodeapp", "helloWorld")
    .setBody(null)
    .setHttpExtension(httpExtension);
  Response response = daprClient.invokeMethod(request, TypeRef.get(Response.class)).block();
  System.out.println("finish getHelloWorld");
  if (response != null) {
    return Response.ok(response.getResult().toString());
  }
  return null;
}

Initialize and test tracing:

kubectl create -f http-demo.yaml
kubectl create -f http-demo-dapr.yaml
curl -H "traceparent: 00-0af7651916cd43dd8438eb211c19319c-b7ad5b7169203331-02" -H "tracestate: congo=t39rcWkgMzE" http://xx.xx.xx.xx:3500/v1.0/invoke/daprnodeapp/method/dapr-helloWorld

Secrets Management: The guide demonstrates creating a Kubernetes secret, configuring a Dapr secret store component, and securely retrieving secrets via the Dapr sidecar API.

kubectl create secret generic dapr-secret --from-literal=my-secret="I'm Batman"

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: mycustomsecretstore
  namespace: default
spec:
  type: secretstores.kubernetes
  version: v1
  metadata:
  - name: "dapr-secret"

Initialize and test secret retrieval:

kubectl create -f daprSecretStore.yaml
curl http://xx.xx.xx.xx:3500/v1.0/secrets/kubernetes/dapr-secret

Configuration Management: Currently in Alpha and gRPC-only, this feature is explored using a Redis-based configuration store. The article provides the component YAML and demonstrates fetching configuration values via the Dapr API.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: configstore
  namespace: default
spec:
  type: configuration.redis
  version: v1
  metadata:
  - name: redisHost
    value: xx.xx.xx.xx:6379
  - name: redisPassword
    value: ''

Initialize and fetch configuration:

kubectl create -f configuration.yaml
curl http://10.246.3.147:3500/v1.0/configuration/configstore/{redisKey}

By completing these hands-on Java demos, developers can effectively integrate Dapr's building blocks into cloud-native microservices architectures.