Building a Replicated Logging System with Apache Kafka

Apache Kafka is a scalable publish-subscribe messaging system
with its core architecture as a distributed commit log.
It was originally built as its centralized event
pipelining platform for online data integration tasks. Over
the past years developing and operating Kafka, we extend
its log-structured architecture as a replicated logging backbone
for much wider application scopes in the distributed
environment. I am going to talk about our design
and engineering experience to replicate Kafka logs for various
distributed data-driven systems, including
source-of-truth data storage and stream processing.

1. Building a Replicated Logging System with Apache Kafka Guozhang Wang, Joel Koshy, Sriram Subramanian, Kartik Paramasivam Mammad Zadeh, Neha Narkhede, Jun Rao, Jay Kreps, Joe Stein Body Level Two Body Level Three Body Level Four Body Level Five

2. We All Love Logs!

3. Apache Kafka A distributed messaging system ..that store messages as a log!

4. Example: LinkedIn back in 2010 Point-to-Point Pipelines What We Want: A Centralized Data Pipeline

5. Log-centric Data Flow Logical Ordering Persistent Buffering “ Source-of-Truth”

6. Store Messages as a Log 4 5 5 7 8 9 10 11 12 ... Producer Write Consumer1 Reads (offset 7) Consumer2 Reads (offset 10) Messages 3

7. Partition the Log across Machines Topic 1 Topic 2 Partitions Producers Producers Consumers Consumers Brokers

8. Apache Kafka Example: Kafka at LinkedIn

9. “Source-of-Truth” should not be lost even when..

10. Replicas and Layout Logs Broker-1 topic1-part1 topic1-part3 topic1-part2 Logs topic1-part2 topic1-part1 topic1-part3 Logs topic1-part3 topic1-part2 topic1-part1 Broker-2 Broker-3

11. Consensus for Log Replication Logs Broker-1 Logs Logs Broker-2 Broker-3 Write Consensus Protocol Consensus Protocol

12. Key Idea Separate membership configuration from data replication

13. Primary-backup Replication Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write

14. Conventional Quorum Commits Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write

15. Conventional Quorum Commits Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write

16. Conventional Quorum Commits Logs Broker-1 * Logs Logs Broker-2 Broker-3

17. Conventional Quorum Commits Logs Broker-1 * Logs Logs Broker-2 Broker-3

18. Leader maintains in-sync-replicas (ISR) Failed / slow follower => drop from ISR Caught-up follower => re-join ISR Producer specifies required ACK based on ISR Configurable ISR Commits

19. Example: ACK with all ISRs Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

20. Example: ACK with all ISRs Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

21. Example: ACK with all ISRs Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

22. Example: ACK with all ISRs Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

23. Example: ACK with all ISRs Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

24. Example: ACK with all ISRs Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

25. Example: ACK with all ISRs Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

26. Example: ACK with Leader-only Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“leader”) ISR { 1, 2, 3 }

27. Example: ACK with Leader-only Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“leader”) ISR { 1, 2, 3 }

28. Example: ACK with Leader-only Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“leader”) ISR { 1, 2, 3 }

29. Example: ACK with Leader-only Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“leader”) ISR { 1, 2, 3 }

30. Example: ACK with Leader-only Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“leader”) ISR { 1, 2, 3 }

31. Example: ACK with Leader-only Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“leader”) ISR { 1, 2, 3 }

32. Example: Slow Follower Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

33. Example: Slow Follower Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

34. Example: Slow Follower Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

35. Example: Slow Follower Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

36. Example: Slow Follower Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2, 3 }

37. Example: Slow Follower Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2 }

38. Example: Slow Follower Logs Broker-1 * Logs Logs Broker-2 Broker-3 Write (ack=“all”) ISR { 1, 2 }

39. Configurable ISR Commits ACK mode Latency On Failures “no" no network delay some data loss “leader" 1 network roundtrip a few data loss “all" ~2 network roundtrips no data loss

40. Use an embedded controller Detect broker failure via ZooKeeper Leader failure => elect new leader from ISR Leader and ISR persisted in Zookeeper For Controller fail-over Membership Management

41. Example: Broker Failure Logs Broker-1 * Logs Logs Broker-2 Broker-3 ISR {1, 2}

42. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 Broker-3

43. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 Broker-3

44. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 Broker-3

45. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 Broker-3 ISR {2 }

46. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 Broker-3 ISR { 2 }

47. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 Broker-3 ISR { 2 }

48. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 * Broker-3 ISR {2}

49. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 * Broker-3 ISR {2}

50. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 * Broker-3 ISR {2}

51. Example: Broker Failure Logs Broker-1 Logs Logs Broker-2 * Broker-3 ISR {2, 3}

52. Overview: Logs and Kafka Log Replication in Kafka Kafka Usage at LinkedIn Conclusion Agenda

53. Change Log Replication

54. Apache Kafka Example: Kafka at LinkedIn

55. Example: Espresso A distributed document store Primary online data serving platform at LI Member profile, homepage, InMail, etc [SIGMOD 2013]

56. Old Espresso Replication Data Center-1 Storage Node Storage Node MySQL Replication MySQL MySQL Search Index Hadoop … … Databus Cross-DC Replicator Data Center-1 Storage Node Storage Node MySQL Replication MySQL MySQL Search Index Hadoop … Databus Cross-DC Replicator

57. Problems with MySQL Replication Master Storage Node P1 Slave Storage Node P2 P3 P4 P5 P6 P1 P2 P3 P4 P5 P6 Binary Log Shipping

58. Replicate Logs with Kafka Storage Node Kafka Logs P1 Storage Node P2 P3 P4 P5 P6 P1 P2 P3 P4 P5 P6 Kafka Producer Kafka Consumer Kafka Consumer Kafka Producer

59. Key-based Log Compaction ... Partition Messages Segment-3 Segment-4 Segment-6 *

60. Key-based Log Compaction d: 3 f: 8 b: 0 c: null ... Partition Messages c: 3 a: 5 a: 6 a: 5 f: 9 ... Segment-3 Segment-4 b: 2 d: 4 a: 1

61. Key-based Log Compaction ... d: 3 f: 8 b: 0 c: null a: 5 f: 9 ... Segment-3 Segment-4 c: 3 a: 5 a: 6 b: 2 d: 4 a: 1 c: 3 a: 5 a: 6 b: 2 d: 4 a: 1 d: 3 f: 8 b: 0 a: 5 f: 9 New Segment Partition Messages

62. Key-based Log Compaction ... d: 3 f: 8 b: 0 c: null a: 5 f: 9 ... Segment-3 Segment-4 c: 3 a: 5 a: 6 b: 2 d: 4 a: 1 c: 3 a: 6 d: 3 f: 8 b: 0 c: null a: 5 f: 9 New Segment Partition Messages

63. Key-based Log Compaction ... d: 3 f: 8 b: 0 c: null a: 5 f: 9 ... Segment-3 Segment-4 c: 3 a: 5 a: 6 b: 2 d: 4 a: 1 d: 3 b: 0 a: 5 f: 9 New Segment Partition Messages

64. Key-based Log Compaction ... d: 3 f: 8 b: 0 c: null a: 5 f: 9 ... Segment-3 Segment-4 c: 3 a: 5 a: 6 b: 2 d: 4 a: 1 d: 3 b: 0 a: 5 f: 9 New Segment Partition Messages

65. New Espresso Replication Data Center-1 Storage Node Storage Node Storage Node Kafka Logs MySQL MySQL MySQL Data Center-n Storage Node Storage Node Storage Node Kafka Logs MySQL MySQL MySQL Kafka MirrorMaker Search Index Hadoop … … Search Index Hadoop … * In Progress

66. Stream Processing

67. Apache Kafka Example: Kafka at LinkedIn

68. Data flow streaming on Kafka and YARN Stateful processing Re-processing Failure Recovery Example: Samza [CIDR 2015]

69. Kafka Kafka Samza State Process Protocol State Process Protocol State Process Protocol Samza Processing

70. Kafka Kafka Samza State Process Protocol State Process Protocol State Process Protocol Samza Processing Kafka Changelog

71. Kafka Kafka Samza State Process Protocol State Process Protocol State Process Protocol Samza Processing Kafka Changlog

72. Kafka Kafka Samza State Process Protocol State Process Protocol State Process Protocol Samza Processing Kafka Changlog State Process Protocol

73. Take-aways Log-centric data flow helps scaling your systems Kafka: replicated log streams for real-time platforms

74. We are Hiring

75. Take-aways Log-centric data flow helps scaling your systems Kafka: replicated log streams for real-time platforms THANKS!