yakkomajuri

Error handling for Kafka engine tables on ClickHouse

Date: 2022-11-08

Objective

Test the failure modes of using the Kafka table error handling approach suggested by this Altinity doc and this ClickHouse PR .

Schemas

yakkomajuri

function atomicsWaitLoop (port : MessagePort, sharedBuffer : Int32Array) {
  if (!useAtomics) return;
  // This function is entered either after receiving the startup message, or
  // when we are done with a task. In those situations, the *only* thing we
  // expect to happen next is a 'message' on `port`.
  // That call would come with the overhead of a C++ → JS boundary crossing,
  // including async tracking. So, instead, if there is no task currently
  // running, we wait for a signal from the parent thread using Atomics.wait(),
  // and read the message from the port instead of generating an event,
  // in order to avoid that overhead.

yakkomajuri

export async function runEveryMinute({ config, cache }) {
    const res = await fetch('https://api.producthunt.com/v2/api/graphql', {
        headers: {
            'Content-Type': 'application/json',
            'Authorization': `Bearer ${config.token}`
        },
        body: JSON.stringify({ 
          "operationName": "GetAllTheData", 
          "variables": {}, 
          "query": "query todayPosts { posts { edges { node { id name tagline votesCount } } } }" }

yakkomajuri

query ProfilePubHandlerQuery($id: ID, $username: ID, $homepagePostsLimit: PaginationLimit, $homepagePostsFrom: String, $includeDistributedResponses: Boolean) {
    userResult(id: $id, username: $username) {
      ... on User {
        id
        name
        username
        bio
        ...ProfilePubScreen_user
      }
    }

yakkomajuri

# clone the repo
git clone https://github.com/yakkomajuri/medium-to-blog
cd medium-to-blog

# substitute your username and URL
export MEDIUM_URL=https://yakkomajuri.medium.com MEDIUM_USERNAME=yakkomajuri 

# run it!
yarn start # or npm start

yakkomajuri

query ProfilePubHandlerQuery($id: ID, $username: ID, $homepagePostsLimit: PaginationLimit, $homepagePostsFrom: String, $includeDistributedResponses: Boolean) {
  userResult(id: $id, username: $username) {
    __typename
    ... on User {
      id
      isFollowing
      name
      username
      viewerIsUser
      ...ProfilePubScreen_user

yakkomajuri

import { kea } from 'kea'
import { counterLogicType } from './counterLogicType'

export const counterLogic = kea<counterLogicType>({
    actions: {
        incrementCounter: true, // https://kea.js.org/docs/guide/concepts#actions
        decrementCounter: true, // true is shorthand for a function that doesn't take any arguments
        updateCounter: (newValue: number) => ({ newValue }),
    },
    reducers: {

yakkomajuri

// webpack.config.js
const path = require('path')
const webpack = require('webpack')

module.exports = {
    entry: './src/index.tsx', // our entry point, as mentioned earlier
    mode: 'development',
    module: {
        rules: [
            {

yakkomajuri

import React, { useState } from 'react'
import { useValues, useActions } from 'kea'
import { counterLogic } from './counterLogic'
import './style.css'

export const Counter = () => {
    const { count } = useValues(counterLogic)
    const { incrementCounter, decrementCounter, updateCounter } = useActions(counterLogic)
    const [inputValue, setInputValue] = useState(0)
    return (

yakkomajuri

{
    "compilerOptions": {
        "baseUrl": "./src",
        "paths": {
            "types/*": ["./types/*"]
        },
        // https://www.sitepoint.com/react-with-typescript-best-practices/
        "allowJs": true, // Allow JavaScript files to be compiled
        "skipLibCheck": true, // Skip type checking of all declaration files
        "esModuleInterop": true, // Disables namespace imports (import * as fs from "fs") and enables CJS/AMD/UMD style imports (import fs from "fs")